ExecutorService，标准的方式避免任务队列获得太满 [英] ExecutorService, standard way to avoid to task queue getting too full

问题描述

我使用ExecutorService来简化并发多线程程序。代码如下：

I am using ExecutorService for ease of concurrent multithreaded program. Take following code:

while(xxx)
 ExecutorService exService = Executors.newFixedThreadPool(NUMBER_THREADS);
 ...  
 Future<..> ... = exService.submit(..);
 ...
}

是，如果所有NUMBER_THREADS都被占用，submit（）不会阻塞。结果是任务队列被许多任务淹没。这样做的后果是，使用ExecutorService.shutdown（）关闭执行服务需要很长时间（ExecutorService.isTerminated（）将为false）。原因是任务队列仍然很满。

In my case the problem is that submit() is not blocking if all NUMBER_THREADS are occupied. The consequence is that the Task queue is getting flooded by many tasks. The consequence of this is, that shutting down the execution service with ExecutorService.shutdown() takes ages (ExecutorService.isTerminated() will be false for long time). Reason is that the task queue is still quite full.

现在我的解决方法是使用信号量，以便不允许在ExecutorService的任务队列中有很多条目： / p>

For now my workaround is to work with semaphores to disallow to have to many entries inside the task queue of ExecutorService:

...
Semaphore semaphore=new Semaphore(NUMBER_THREADS);

while(xxx)
 ExecutorService exService = Executors.newFixedThreadPool(NUMBER_THREADS); 
 ...
 semaphore.aquire();  
 // internally the task calls a finish callback, which invokes semaphore.release()
 // -> now another task is added to queue
 Future<..> ... = exService.submit(..); 
 ...
}

一个更好的封装解决方案？

I am sure there is a better more encapsulated solution?

推荐答案

诀窍是使用固定的队列大小和：

The trick is to use a fixed queue size and:

new ThreadPoolExecutor.CallerRunsPolicy()

我还建议使用Guava的 ListeningExecutorService 。
这是一个示例消费者/生产者队列。

I also recommend using Guava's ListeningExecutorService. Here is an example consumer/producer queues.

private ListeningExecutorService producerExecutorService = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
private ListeningExecutorService consumerExecutorService = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));

private static ExecutorService newFixedThreadPoolWithQueueSize(int nThreads, int queueSize) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  5000L, TimeUnit.MILLISECONDS,
                                  new ArrayBlockingQueue<Runnable>(queueSize, true), new ThreadPoolExecutor.CallerRunsPolicy());
}

任何更好的，你可能想考虑一个MQ像RabbitMQ或ActiveMQ有QoS技术。

Anything better and you might want to consider a MQ like RabbitMQ or ActiveMQ as they have QoS technology.

这篇关于ExecutorService，标准的方式避免任务队列获得太满的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！