ExecutorService, cómo esperar a que finalicen todas las tareas

¿Cuál es la forma más sencilla de esperar ExecutorServicea que finalicen todas las tareas ? Mi tarea es principalmente computacional, por lo que solo quiero ejecutar una gran cantidad de trabajos, uno en cada núcleo. En este momento mi configuración se ve así:

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases) {   
    es.execute(new ComputeDTask(singleTable));
}
try{
    es.wait();
} 
catch (InterruptedException e){
    e.printStackTrace();
}

ComputeDTaskImplementos ejecutables. Esto aparece para ejecutar las tareas correctamente, pero el código se bloquea en wait()la IllegalMonitorStateException. Esto es extraño, porque jugué con algunos ejemplos de juguetes y parecía funcionar.

uniquePhrasescontiene varias decenas de miles de elementos. ¿Debo estar usando otro método? Estoy buscando algo lo mas simple posible

El enfoque más simple es usar el ExecutorService.invokeAll()que hace lo que quieres en una sola línea. En su lenguaje, necesitará modificar o ajustar ComputeDTaskpara implementar Callable<>, lo que puede brindarle un poco más de flexibilidad. Probablemente en su aplicación hay una implementación significativa de Callable.call(), pero aquí hay una manera de envolverla si no la está usando Executors.callable().

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

Como otros han señalado, puede usar la versión de tiempo de espera de, invokeAll()si corresponde. En este ejemplo, answersva a contener un montón de Futures que devolverán valores nulos (consulte la definición de Executors.callable(). Probablemente lo que quiera hacer es una ligera refactorización para que pueda obtener una respuesta útil o una referencia al subyacente ComputeDTask, pero puedo No cuentes con tu ejemplo.

Si no está claro, tenga en cuenta que invokeAll()no volverá hasta que se completen todas las tareas. (es decir, todos los mensajes de correo electrónico de Futuresu answerscolección informarán .isDone()si se le solicita). Esto evita todo el apagado manual, esperar la terminación, etc. y le permite volver a usarlo ExecutorServiceordenadamente durante varios ciclos, si lo desea.

Hay algunas preguntas relacionadas sobre SO:

• Cómo esperar a que todos los hilos terminen

• Valores de retorno de hilos Java

• invokeAll () no está dispuesto a aceptar una Colección <invocable <t>>

• ¿Necesito sincronizar?

Ninguno de estos es estrictamente adecuado para su pregunta, pero proporcionan un poco de color sobre cómo la gente piensa Executor/ ExecutorServicedebe ser utilizado.

Si desea esperar a que se completen todas las tareas, use el shutdownmétodo en lugar de wait. Luego sígalo con awaitTermination.

Además, puede usar Runtime.availableProcessorspara obtener el número de subprocesos de hardware para que pueda inicializar su conjunto de subprocesos correctamente.

Si esperar ExecutorServicea que finalicen todas las tareas no es precisamente su objetivo, sino esperar hasta que se haya completado un lote específico de tareas , puede utilizar un CompletionService- específicamente, un ExecutorCompletionService.

La idea es crear una ExecutorCompletionServiceenvoltura Executor, enviar una cantidad conocida de tareas a través de CompletionService, y luego extraer esa misma cantidad de resultados de la cola de finalización utilizando take()(qué bloques) o poll()(qué no). Una vez que haya dibujado todos los resultados esperados correspondientes a las tareas que envió, sabrá que todos están listos.

Permítanme decir esto una vez más, porque no es obvio desde la interfaz: debe saber cuántas cosas pone en el CompletionServicepara poder saber cuántas cosas tratar de extraer. Esto importa especialmente con el take()método: llámelo una vez más y bloqueará su hilo de llamada hasta que otro hilo envíe otro trabajo al mismo CompletionService.

Hay algunos ejemplos que muestran cómo usarCompletionService en el libro Java Concurrency in Practice .

Si desea esperar a que el servicio del ejecutor termine de ejecutarse, llame shutdown()y luego, espere Terminación (unidades, tipo de unidad) , por ejemplo awaitTermination(1, MINUTE). El ExecutorService no bloquea en su propio monitor, por lo que no puede usar, waitetc.

Podría esperar a que finalicen los trabajos en un intervalo determinado:

int maxSecondsPerComputeDTask = 20;
try {
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS)) {
        // consider giving up with a 'break' statement under certain conditions
    }
} catch (InterruptedException e) {
    throw new RuntimeException(e);    
}

O puede usar ExecutorService . submit ( Runnable ) y recoge los objetos Future que devuelve y llama a get () en cada turno para esperar a que finalicen.

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases) {
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures) {
   try {
       future.get();
   } catch (InterruptedException e) {
       throw new RuntimeException(e);
   } catch (ExecutionException e) {
       throw new RuntimeException(e);
   }
}

InterruptedException es extremadamente importante para manejar adecuadamente. Es lo que le permite a usted o a los usuarios de su biblioteca finalizar un proceso largo de forma segura.

Solo usa

latch = new CountDownLatch(noThreads)

En cada hilo

latch.countDown();

y como barrera

latch.await();

Causa raíz de IllegalMonitorStateException :

Lanzado para indicar que un subproceso ha intentado esperar en el monitor de un objeto o para notificar a otros subprocesos que esperan en el monitor de un objeto sin poseer el monitor especificado.

Desde su código, acaba de llamar a wait () en ExecutorService sin poseer el bloqueo.

Debajo del código arreglará IllegalMonitorStateException

try 
{
    synchronized(es){
        es.wait(); // Add some condition before you call wait()
    }
} 

Siga uno de los siguientes enfoques para esperar la finalización de todas las tareas que se han enviado ExecutorService.

1. Itere a través de todas las Futuretareas desde submitencendido ExecutorServicey verifique el estado bloqueando la llamada get()en el Futureobjeto

2. Usando invokeAll enExecutorService

3. Usando CountDownLatch

4. Usando ForkJoinPool o newWorkStealingPool de Executors(desde java 8)

5. Cierre el grupo como se recomienda en la página de documentación de Oracle

   void shutdownAndAwaitTermination(ExecutorService pool) {
      pool.shutdown(); // Disable new tasks from being submitted
      try {
      // Wait a while for existing tasks to terminate
      if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
          pool.shutdownNow(); // Cancel currently executing tasks
          // Wait a while for tasks to respond to being cancelled
          if (!pool.awaitTermination(60, TimeUnit.SECONDS))
          System.err.println("Pool did not terminate");
      }
   } catch (InterruptedException ie) {
        // (Re-)Cancel if current thread also interrupted
        pool.shutdownNow();
        // Preserve interrupt status
        Thread.currentThread().interrupt();
   }

   Si desea esperar con gracia la finalización de todas las tareas cuando utiliza la opción 5 en lugar de las opciones 1 a 4, cambie

   if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {

   a

   un while(condition)que verifica cada 1 minuto.

Puede usar el ExecutorService.invokeAllmétodo, ejecutará todas las tareas y esperará hasta que todos los hilos terminen su tarea.

Aquí está completo javadoc

También puede usar una versión sobrecargada de este método para especificar el tiempo de espera.

Aquí hay un código de muestra con ExecutorService.invokeAll

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newFixedThreadPool(3);
        List<Callable<String>> taskList = new ArrayList<>();
        taskList.add(new Task1());
        taskList.add(new Task2());
        List<Future<String>> results = service.invokeAll(taskList);
        for (Future<String> f : results) {
            System.out.println(f.get());
        }
    }

}

class Task1 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(2000);
            return "Task 1 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task1";
        }
    }
}

class Task2 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(3000);
            return "Task 2 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task2";
        }
    }
}

También tengo la situación de que tengo que rastrear un conjunto de documentos. Comienzo con un documento inicial "inicial" que debe procesarse, ese documento contiene enlaces a otros documentos que también deben procesarse, y así sucesivamente.

En mi programa principal, solo quiero escribir algo como lo siguiente, donde Crawlercontrola un montón de hilos.

Crawler c = new Crawler();
c.schedule(seedDocument); 
c.waitUntilCompletion()

La misma situación sucedería si quisiera navegar por un árbol; aparecería en el nodo raíz, el procesador para cada nodo agregaría hijos a la cola según sea necesario, y un grupo de subprocesos procesaría todos los nodos en el árbol, hasta que no hubiera más.

No pude encontrar nada en la JVM que me pareció un poco sorprendente. Así que escribí una clase ThreadPoolque se puede usar directamente o subclase para agregar métodos adecuados para el dominio, por ejemplo schedule(Document). ¡Espero eso ayude!

ThreadPool Javadoc | Maven

Agregue todos los hilos en la colección y envíelo usando invokeAll. Si puede usar el invokeAllmétodo de ExecutorService, JVM no procederá a la siguiente línea hasta que todos los hilos estén completos.

Aquí hay un buen ejemplo: invokeAll a través de ExecutorService

Envíe sus tareas al Runner y luego espere llamando al método waitTillDone () de esta manera:

Runner runner = Runner.runner(2);

for (DataTable singleTable : uniquePhrases) {

    runner.run(new ComputeDTask(singleTable));
}

// blocks until all tasks are finished (or failed)
runner.waitTillDone();

runner.shutdown();

Para usarlo agregue esta dependencia gradle / maven: 'com.github.matejtymes:javafixes:1.0'

Para obtener más detalles, consulte aquí: https://github.com/MatejTymes/JavaFixes o aquí: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

Una alternativa simple a esto es usar hilos junto con join. Consulte: Unir hilos

Solo esperaré a que el ejecutor termine con un tiempo de espera especificado que crees que es adecuado para completar las tareas.

 try {  
         //do stuff here 
         exe.execute(thread);
    } finally {
        exe.shutdown();
    }
    boolean result = exe.awaitTermination(4, TimeUnit.HOURS);
    if (!result)

    {
        LOGGER.error("It took more than 4 hour for the executor to stop, this shouldn't be the normal behaviour.");
    }

Suena como lo que necesita ForkJoinPooly utiliza el grupo global para ejecutar tareas.

public static void main(String[] args) {
    // the default `commonPool` should be sufficient for many cases.
    ForkJoinPool pool = ForkJoinPool.commonPool(); 
    // The root of your task that may spawn other tasks. 
    // Make sure it submits the additional tasks to the same executor that it is in.
    Runnable rootTask = new YourTask(pool); 
    pool.execute(rootTask);
    pool.awaitQuiescence(...);
    // that's it.
}

La belleza está en pool.awaitQuiescencedonde el método bloqueará, utilice el hilo de la persona que llama para ejecutar sus tareas y luego regrese cuando esté realmente vacío.