Как да изчакам всички задачи в 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 за управление на партиди задачи, вижте също това отговор.

(Това е опит да възпроизведа по-ранния, изтрит отговор на Thilo с мои собствени корекции.)

Мисля, че може да се наложи да изясните въпроса си, тъй като има имплицитно безкрайно условие... в един момент трябва да решите да изключите своя изпълнител и в този момент той няма да приема повече задачи. Вашият въпрос изглежда предполага, че искате да изчакате, докато разберете, че няма да бъдат изпращани други задачи, което можете да знаете само в собствения си код на приложение.

Следният отговор ще ви позволи плавно да преминете към нов TPE (по каквато и да е причина), като завършите всички подадени в момента задачи и не отхвърляте нови задачи към новия TPE. Може да отговори на въпроса ви. @Thilo може също.

Ако приемем, че сте дефинирали някъде видим 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) {
            }
        }