ExecutorService, moyen standard pour éviter que la file d'attente des tâches ne soit trop pleine

Question

J'utilise ExecutorService pour faciliter le programme multithread concurrent. Prenez le code suivant:

 

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

Dans mon cas, le problème est que submit() ne bloque pas si tous les NUMBER_THREADS sont occupés. La conséquence est que la file d'attente des tâches est inondée par de nombreuses tâches. La conséquence de ceci est que l'arrêt du service d'exécution avec ExecutorService.shutdown() prend des âges (ExecutorService.isTerminated() sera faux pendant longtemps). Raison est que la file d'attente de tâches est encore assez complète.

Pour l'instant ma solution de contournement est de travailler avec des sémaphores pour désavouer avoir à plusieurs entrées dans la file d'attente des tâches de 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(..); 
... 
} 
 

Je suis sûr qu'il ya une meilleure solution plus encapsulées?

Answer 1

Il est préférable de créer le ThreadPoolExecutor vous-même (ce que fait Executors.newXXX() de toute façon).

Dans le constructeur, vous pouvez passer une BlockingQueue à utiliser par l'Executor comme file d'attente de tâches. Si vous passez une taille bloquée BlockingQueue (comme LinkedBlockingQueue), il devrait obtenir l'effet que vous voulez.

ExecutorService exService = new ThreadPoolExecutor(NUMBER_THREADS, NUMBER_THREADS, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(workQueueSize)); 

Answer 2

Vous pouvez ThreadPoolExecutor.getQueue() taille.() Pour connaître la taille de la file d'attente. Vous pouvez effectuer une action si la file d'attente est trop longue. Je suggère d'exécuter la tâche dans le thread actuel si la file d'attente est trop longue pour ralentir le producteur (si cela est approprié)

Answer 3

Un vrai blocage ThreadPoolExecutor a été sur la liste de souhaits de beaucoup, il y a même un bogue JDC ouvert dessus . Je suis face au même problème, et suis tombé sur ceci: http://today.java.net/pub/a/today/2008/10/23/creating-a-notifying-blocking-thread-pool-executor.html

Il est une mise en œuvre d'un BlockingThreadPoolExecutor, mis en œuvre au moyen d'un RejectionPolicy qui utilise l'offre pour ajouter la tâche à la file d'attente, en attendant que la file d'attente d'avoir de la place. Ça à l'air bon.

Answer 4

Vous pouvez ajouter une autre file d'attente bloquante qui a une taille limitée pour contrôler la taille de la file d'attente interne dans executorService, certains pensent comme un sémaphore mais très facile. avant l'exécuteur que vous mettez() et quand la tâche achive take(). prendre() doit être à l'intérieur du code de tâche

Answer 5

L'astuce consiste à utiliser une taille de file d'attente fixe et:

new ThreadPoolExecutor.CallerRunsPolicy() 

Je recommande également d'utiliser ListeningExecutorService de goyave. Voici un exemple de files d'attente consommateur/producteur.

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()); 
} 

quelque chose de mieux et vous voudrez peut-être envisager un MQ comme RabbitMQ ou ActiveMQ car ils ont la technologie QoS.