Как дождаться завершения всех задач в ThreadPoolExecutor, не завершая работу Executor?

Если вам интересно знать, когда завершается определенная задача или определенный пакет задач, вы можете использовать ExecutorService.submit(Runnable). Вызов этого метода возвращает объект Future, который может быть помещен в Collection, который ваш основной поток затем будет перебирать, вызывая Future.get() для каждого из них. Это приведет к тому, что ваш основной поток остановит выполнение, пока ExecutorService не обработает все Runnable задачи.

Collection<Future<?>> futures = new LinkedList<Future<?>>();
futures.add(executorService.submit(myRunnable));
for (Future<?> future:futures) {
    future.get();
}

My Scenario - это поисковый робот, который получает некоторую информацию с веб-сайта и затем обрабатывает ее. ThreadPoolExecutor используется для ускорения процесса, потому что за это время может быть загружено много страниц. Таким образом, новые задачи будут созданы в существующей задаче, потому что поисковый робот будет переходить по гиперссылкам на каждой странице. Проблема та же: основной поток не знает, когда все задачи выполнены, и может начать обрабатывать результат. Я использую простой способ определить это. Это не очень элегантно, но в моем случае работает:

while (executor.getTaskCount()!=executor.getCompletedTaskCount()){
    System.err.println("count="+executor.getTaskCount()+","+executor.getCompletedTaskCount());
    Thread.sleep(5000);
}
executor.shutdown();
executor.awaitTermination(60, TimeUnit.SECONDS);

Возможно, вы ищете CompletionService для управления пакетами задач см. также это ответ.

(Это попытка воспроизвести ранее удаленный ответ Тило с моими собственными корректировками.)

Я думаю, вам может потребоваться прояснить свой вопрос, поскольку существует неявное бесконечное условие ... в какой-то момент вы должны решить закрыть своего исполнителя, и в этот момент он больше не будет принимать никаких задач. Похоже, ваш вопрос подразумевает, что вы хотите подождать, пока вы узнаете, что никакие дальнейшие задачи не будут отправлены, о чем вы можете узнать только в своем собственном коде приложения.

Следующий ответ позволит вам плавно перейти на новый TPE (по любой причине), выполнив все текущие отправленные задачи и не отклоняя новые задачи для нового TPE. Это может ответить на ваш вопрос. @ Могущество Тило тоже.

Предполагая, что вы где-то определили видимый TPE, который используется как таковой:

AtomicReference<ThreadPoolExecutor> publiclyAvailableTPE = ...;

Затем вы можете написать процедуру замены TPE как таковую. Его также можно было бы написать с помощью синхронизированного метода, но я думаю, что это проще:

void rotateTPE()
{
   ThreadPoolExecutor newTPE = createNewTPE();
   // atomic swap with publicly-visible TPE
   ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(newTPE);
   oldTPE.shutdown();

   // and if you want this method to block awaiting completion of old tasks in  
   // the previously visible TPE
   oldTPE.awaitTermination();
} 

В качестве альтернативы, если вы действительно не шутите, хотите убить пул потоков, тогда ваша сторона отправителя должна будет справиться с отклоненными задачами в какой-то момент, и вы можете использовать null для нового TPE:

void killTPE()
{
   ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(null);
   oldTPE.shutdown();

   // and if you want this method to block awaiting completion of old tasks in  
   // the previously visible TPE
   oldTPE.awaitTermination();
} 

Что может вызвать проблемы с восходящим потоком, вызывающий должен знать, что делать с null.

Вы также можете заменить фиктивный TPE, который просто отклоняет каждое новое выполнение, но это эквивалентно тому, что происходит, если вы вызываете shutdown() на TPE.

Если вы не хотите использовать shutdown, выполните следующие действия:

1. Выполните итерации по всем Future задачам из отправки на ExecutorService и проверьте статус с блокирующим вызовом get() на Future объекте, как предложено Tim Bender

2. Используйте один из

   1. Using invokeAll on ExecutorService
   2. Использование CountDownLatch
   3. Использование ForkJoinPool или newWorkStealingPool из Executors (с Java 8)

invokeAll() на службе исполнителя также достигает той же цели, что и CountDownLatch

Связанный вопрос SE:

Как дождаться нескольких потоков завершить?

Вы можете вызвать waitTillDone () в классе Runner:

Runner runner = Runner.runner(10);

runner.runIn(2, SECONDS, runnable);
runner.run(runnable); // each of this runnables could submit more tasks

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

// and now reuse it

runner.runIn(500, MILLISECONDS, callable);

runner.waitTillDone();
runner.shutdown();

Чтобы использовать его, добавьте в свой проект эту зависимость gradle / maven: 'com.github.matejtymes:javafixes:1.0'

Дополнительные сведения см. Здесь: https://github.com/MatejTymes/JavaFixes или здесь: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

Попробуйте использовать размер очереди и количество активных задач, как показано ниже.

 while (executor.getThreadPoolExecutor().getActiveCount() != 0 || !executor.getThreadPoolExecutor().getQueue().isEmpty()){
                     try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
            }
        }