ExecutorService, como aguardar a conclusão de todas as tarefas

Qual é a maneira mais simples de aguardar o ExecutorServicetérmino de todas as tarefas ? Minha tarefa é principalmente computacional, então eu só quero executar um grande número de tarefas - uma em cada núcleo. No momento, minha configuração fica assim:

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

ComputeDTaskimplementa executável. Isso parece executar as tarefas corretamente, mas o código falha wait()com IllegalMonitorStateException. Isso é estranho, porque eu brinquei com alguns exemplos de brinquedos e pareceu funcionar.

uniquePhrasescontém várias dezenas de milhares de elementos. Devo estar usando outro método? Estou procurando algo o mais simples possível

A abordagem mais simples é usar ExecutorService.invokeAll()o que você deseja em uma única linha. Na sua linguagem, você precisará modificar ou agrupar ComputeDTaskpara implementar Callable<>, o que pode lhe dar um pouco mais de flexibilidade. Provavelmente em seu aplicativo há uma implementação significativa de Callable.call(), mas aqui está uma maneira de envolvê-lo, se não estiver 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 outros já apontaram, você pode usar a versão de tempo limite, invokeAll()se apropriado. Neste exemplo, answersirá conter um monte de Futures que retornará nulos (consulte a definição de Executors.callable(). Provavelmente o que você deseja fazer é uma leve refatoração para que você possa obter uma resposta útil de volta ou uma referência ao subjacente ComputeDTask, mas eu posso conte do seu exemplo.

Se não estiver claro, observe que invokeAll()não retornará até que todas as tarefas sejam concluídas. (ou seja, todos os Futures da sua answerscoleção reportarão .isDone()se solicitado.) Isso evita todo o desligamento manual, aguarda o término, etc ... e permite reutilizá-lo ExecutorServiceordenadamente por vários ciclos, se desejado.

Existem algumas perguntas relacionadas ao SO:

• Como aguardar a conclusão de todos os threads

• Retornar valores dos encadeamentos java

• invokeAll () não está disposto a aceitar uma coleção <Callable <t>>

• Preciso sincronizar?

Nada disso é estritamente correto para sua pergunta, mas eles fornecem um pouco de cor sobre como as pessoas pensam Executor/ ExecutorServicedevem ser usadas.

Se você deseja aguardar a conclusão de todas as tarefas, use o shutdownmétodo em vez de wait. Então siga com awaitTermination.

Além disso, você pode usar Runtime.availableProcessorspara obter o número de threads de hardware para inicializar seu pool de threads corretamente.

Se aguardar ExecutorServicea conclusão de todas as tarefas não for exatamente o seu objetivo, mas aguardar até que um lote específico de tarefas seja concluído, você poderá usar um CompletionService- especificamente, um ExecutorCompletionService.

A idéia é criar um ExecutorCompletionServiceempacotamento Executor, enviar um número conhecido de tarefas através do e CompletionService, em seguida, desenhar o mesmo número de resultados da fila de conclusão usando take()(quais blocos) oupoll() (quais não). Depois de desenhar todos os resultados esperados correspondentes às tarefas enviadas, você sabe que todas elas foram concluídas.

Deixe-me declarar isso mais uma vez, porque não é óbvio a partir da interface: você deve saber quantas coisas você colocou CompletionServicepara saber quantas coisas tentar extrair. Isso é importante principalmente com o take()método: chame-o muitas vezes e ele bloqueará seu thread de chamada até que outro thread envie outro trabalho para o mesmo CompletionService.

Existem alguns exemplos que mostram como usarCompletionService no livro Java Concurrency in Practice .

Se você deseja esperar o serviço executor terminar a execução, chame shutdown()e aguardeTermination (units, unitType) , por exemplo awaitTermination(1, MINUTE). O ExecutorService não bloqueia em seu próprio monitor, portanto você não pode usar waitetc.

Você pode esperar a conclusão dos trabalhos em um determinado intervalo:

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

Ou você pode usar o ExecutorService . envie ( Runnable ) e colete os objetos Future que ele retorna e chame get () em cada um deles para aguardar o término.

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 é extremamente importante para lidar adequadamente. É o que permite que você ou os usuários da sua biblioteca encerrem um longo processo com segurança.

Apenas use

latch = new CountDownLatch(noThreads)

Em cada thread

latch.countDown();

e como barreira

latch.await();

Causa raiz de IllegalMonitorStateException :

Lançado para indicar que um encadeamento tentou aguardar no monitor de um objeto ou para notificar outros encadeamentos que aguardam no monitor de um objeto sem possuir o monitor especificado.

No seu código, você acabou de chamar wait () no ExecutorService sem possuir o bloqueio.

O código abaixo será corrigido IllegalMonitorStateException

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

Siga uma das abordagens abaixo para aguardar a conclusão de todas as tarefas submetidas ExecutorService.

1. Repita todas as Futuretarefas a partir submitde ExecutorServicee verifique o status com o bloqueio de chamadas get()no Futureobjeto

2. Usando invokeAll onExecutorService

3. Usando CountDownLatch

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

5. Encerre o pool conforme recomendado na página de documentação da 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();
   }

   Se você desejar esperar por todas as tarefas para conclusão quando estiver usando a opção 5 em vez das opções 1 a 4, altere

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

   para

   um while(condition)que verifica a cada 1 minuto.

Você pode usar o ExecutorService.invokeAllmétodo, ele executará todas as tarefas e esperará até que todos os threads concluam sua tarefa.

Aqui está o javadoc completo

Você também pode usar a versão sobrecarregada desse método para especificar o tempo limite.

Aqui está um código de exemplo com 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";
        }
    }
}

Também tenho a situação de ter um conjunto de documentos a serem rastreados. Começo com um documento inicial "inicial" que deve ser processado, que contém links para outros documentos que também devem ser processados ​​e assim por diante.

No meu programa principal, eu só quero escrever algo como o seguinte, onde Crawlercontrola um monte de threads.

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

A mesma situação aconteceria se eu quisesse navegar em uma árvore; eu apareceria no nó raiz, o processador de cada nó adicionaria filhos à fila, conforme necessário, e um monte de threads processaria todos os nós na árvore, até que não houvesse mais.

Não consegui encontrar nada na JVM que achei um pouco surpreendente. Então, eu escrevi uma classe ThreadPoolque pode ser usada diretamente ou subclasse para adicionar métodos adequados para o domínio, por exemplo schedule(Document). Espero que ajude!

ThreadPool Javadoc | Maven

Adicione todos os tópicos da coleção e envie-os usando invokeAll. Se você puder usar o invokeAllmétodo of ExecutorService, a JVM não continuará para a próxima linha até que todos os encadeamentos sejam concluídos.

Aqui está um bom exemplo: invokeAll via ExecutorService

Envie suas tarefas para o Runner e aguarde chamando o método waitTillDone () assim:

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 usá-lo, adicione esta dependência gradle / maven: 'com.github.matejtymes:javafixes:1.0'

Para obter mais detalhes, consulte aqui: https://github.com/MatejTymes/JavaFixes ou aqui: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

Uma alternativa simples para isso é usar threads junto com join. Consulte: Juntando tópicos

Vou aguardar que o executor termine com um tempo limite especificado que você acha que é adequado para a conclusão das tarefas.

 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.");
    }

Parece que você precisa ForkJoinPoole usa o pool global para executar tarefas.

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.
}

A beleza é pool.awaitQuiescenceonde o método bloqueará utilizar o encadeamento do chamador para executar suas tarefas e retornar quando estiver realmente vazio.