How do you limit threads in an ExecutorService?

Question

I use ExecutorService to run many tasks in different threads. Sometimes, too many Runnable instances waiting in a thread pool can cause a memory problem.

I am trying to write an executor of a blocking task to solve it. Is there a formal solution for this?

For instance:

BlockingJobExecutor executor = new BlockingJobExecutor(3); for (int i = 0; i < 1000; i++) { executor.addJob(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } LogFactory.getLog(BTest.class).info("test " + System.currentTimeMillis()); } }); } executor.shutdown(); 

here is the BlockingJobExecutor class:

 import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class BlockingJobExecutor { AtomicInteger counter = new AtomicInteger(); ExecutorService service; int threads; public BlockingJobExecutor(int threads) { if (threads < 1) { throw new IllegalArgumentException("threads must be greater than 1."); } service = Executors.newFixedThreadPool(threads); this.threads = threads; } static class JobWrapper implements Runnable { BlockingJobExecutor executor; Runnable job; public JobWrapper(BlockingJobExecutor executor, Runnable job) throws InterruptedException { synchronized (executor.counter) { while (executor.counter.get() >= executor.limit()) { executor.counter.wait(); } } this.executor = executor; this.job = job; } @Override public void run() { try { job.run(); } finally { synchronized (executor.counter) { executor.counter.decrementAndGet(); executor.counter.notifyAll(); } } } } public int limit() { return threads; } public void shutdown() { service.shutdown(); try { service.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { throw new RuntimeException(e); } } public void addJob(Runnable job) { try { service.execute(new JobWrapper(this, job)); } catch (InterruptedException e) { throw new RuntimeException(e); } } } 

Answer 1

There are two ways this can happen. You may have too many running pending queues, or too many threads running simultaneously. If there are too many jobs in the queue, you can use the fixed size BlockingQueue in the ExecutorService to limit the number of items that can be queued. Then, when you try to queue a new task, the operation will be blocked until there is no space in the queue.

If too many threads are running, you can limit the number of threads available to run tasks in the ExecutorService by calling Executors.newFixedThreadPool with the number of threads required.