resueltos - java multitarea

ExecutorService, forma estándar para evitar que la cola de tareas se llene demasiado

Question

Estoy utilizando ExecutorService para facilitar el programa concurrente multiproceso. Toma el siguiente código:

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

En mi caso, el problema es que submit () no está bloqueando si todos los NUMBER_THREADS están ocupados. La consecuencia es que la cola de Tareas se inunda con muchas tareas. La consecuencia de esto es que el cierre del servicio de ejecución con ExecutorService.shutdown () lleva años (ExecutorService.isTerminated () será falso durante mucho tiempo). La razón es que la cola de tareas todavía está bastante llena.

Por ahora mi solución es trabajar con semáforos para no permitir tener muchas entradas dentro de la cola de tareas 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(..); 
 ...
}

Estoy seguro de que hay una mejor solución más encapsulada?

Answer 1

El truco es usar un tamaño de cola fijo y:

new ThreadPoolExecutor.CallerRunsPolicy()

También recomiendo usar el ListeningExecutorService de Guava. Aquí hay un ejemplo de colas consumidor / productor.

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

Algo mejor y es posible que desee considerar un MQ como RabbitMQ o ActiveMQ ya que tienen tecnología QoS.

Answer 2

Es mejor que crees ThreadPoolExecutor tu mismo (que es lo que Executors.newXXX () hace de todos modos).

En el constructor, puede pasar un BlockingQueue para que lo ejecute el Ejecutor como su cola de tareas. Si transfiere un BlockingQueue con restricciones de tamaño (como LinkedBlockingQueue ), debe lograr el efecto que desee.

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

Answer 3

Puede ThreadPoolExecutor.getQueue (). Size () para averiguar el tamaño de la cola de espera. Puede realizar una acción si la cola es demasiado larga. Sugiero ejecutar la tarea en el hilo actual si la cola es demasiado larga para ralentizar al productor (si es apropiado)

Answer 4

Un verdadero bloqueador ThreadPoolExecutor ha estado en la lista de deseos de muchos, incluso hay un error JDC abierto en él. Estoy enfrentando el mismo problema, y ​​me encontré con esto: http://today.java.net/pub/a/today/2008/10/23/creating-a-notifying-blocking-thread-pool-executor.html

Es una implementación de un BlockingThreadPoolExecutor, implementado usando una RejectionPolicy que usa la oferta para agregar la tarea a la cola, esperando que la cola tenga espacio. Se ve bien.

Answer 5

puede agregar otra cola de bloquing que tenga un tamaño limitado para controlar el tamaño de la cola interna en executorService, algunos piensan como semáforos pero es muy fácil. antes del ejecutor, pones () y cuando la tarea se completa, toma (). take () debe estar dentro del código de la tarea