ExecutorService - Esperando a que finalicen los subprocesos

1. Información general

El marco ExecutorService facilita el procesamiento de tareas en varios subprocesos. Vamos a ejemplificar algunos escenarios en los que esperamos que los subprocesos terminen su ejecución.

Además, mostraremos cómo cerrar correctamente un ExecutorService y esperar a que los subprocesos que ya están en ejecución terminen su ejecución.

2. Después del cierre del ejecutor

Cuando usamos un Ejecutor, podemos cerrarlo llamando a los métodos shutdown () o shutdownNow () . Sin embargo, no esperará hasta que todos los hilos dejen de ejecutarse.

Se puede esperar a que los subprocesos existentes completen su ejecución mediante el método awaitTermination () .

Esto bloquea el hilo hasta que todas las tareas completen su ejecución o se alcance el tiempo de espera especificado:

public void awaitTerminationAfterShutdown(ExecutorService threadPool) { threadPool.shutdown(); try { if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { threadPool.shutdownNow(); } } catch (InterruptedException ex) { threadPool.shutdownNow(); Thread.currentThread().interrupt(); } }

3. Usando CountDownLatch

A continuación, veamos otro enfoque para resolver este problema: usar un CountDownLatch para señalar la finalización de una tarea.

Podemos inicializarlo con un valor que represente la cantidad de veces que se puede disminuir antes de que se notifique a todos los subprocesos que han llamado al método await () .

Por ejemplo, si necesitamos que el hilo actual espere a que otros N hilos terminen su ejecución, podemos inicializar el pestillo usando N :

ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(10); CountDownLatch latch = new CountDownLatch(2); for (int i = 0; i  { try { // ... latch.countDown(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // wait for the latch to be decremented by the two remaining threads latch.await();

4. Usando invokeAll ()

El primer enfoque que podemos usar para ejecutar subprocesos es el método invokeAll () . El método devuelve una lista de objetos futuros después de que finalizan todas las tareas o expira el tiempo de espera .

Además, debemos tener en cuenta que el orden de los objetos Future devueltos es el mismo que la lista de los objetos invocables proporcionados :

ExecutorService WORKER_THREAD_POOL = Executors.newFixedThreadPool(10); List
    
      callables = Arrays.asList( new DelayedCallable("fast thread", 100), new DelayedCallable("slow thread", 3000)); long startProcessingTime = System.currentTimeMillis(); List
     
       futures = WORKER_THREAD_POOL.invokeAll(callables); awaitTerminationAfterShutdown(WORKER_THREAD_POOL); long totalProcessingTime = System.currentTimeMillis() - startProcessingTime; assertTrue(totalProcessingTime >= 3000); String firstThreadResponse = futures.get(0).get(); assertTrue("fast thread".equals(firstThreadResponse)); String secondThreadResponse = futures.get(1).get(); assertTrue("slow thread".equals(secondThreadResponse));
     
    

5. Uso de ExecutorCompletionService

Otro enfoque para ejecutar varios subprocesos es mediante ExecutorCompletionService. Utiliza un ExecutorService proporcionado para ejecutar tareas.

Una diferencia sobre invokeAll () es el orden en el que se devuelven los futuros, que representan las tareas ejecutadas. ExecutorCompletionService usa una cola para almacenar los resultados en el orden en que terminaron , mientras que invokeAll () devuelve una lista con el mismo orden secuencial que el iterador para la lista de tareas dada:

CompletionService service = new ExecutorCompletionService(WORKER_THREAD_POOL); List
    
      callables = Arrays.asList( new DelayedCallable("fast thread", 100), new DelayedCallable("slow thread", 3000)); for (Callable callable : callables) { service.submit(callable); } 
    

Se puede acceder a los resultados usando el método take () :

long startProcessingTime = System.currentTimeMillis(); Future future = service.take(); String firstThreadResponse = future.get(); long totalProcessingTime = System.currentTimeMillis() - startProcessingTime; assertTrue("First response should be from the fast thread", "fast thread".equals(firstThreadResponse)); assertTrue(totalProcessingTime >= 100 && totalProcessingTime = 3000 && totalProcessingTime < 4000); LOG.debug("Thread finished after: " + totalProcessingTime + " milliseconds"); awaitTerminationAfterShutdown(WORKER_THREAD_POOL);

6. Conclusión

Dependiendo del caso de uso, tenemos varias opciones para esperar a que los subprocesos terminen su ejecución.

Un CountDownLatch es útil cuando necesitamos un mecanismo para notificar a uno o más subprocesos que un conjunto de operaciones realizadas por otros subprocesos ha finalizado.

ExecutorCompletionService es útil cuando necesitamos acceder al resultado de la tarea lo antes posible y otros enfoques cuando queremos esperar a que finalicen todas las tareas en ejecución.

El código fuente del artículo está disponible en GitHub.