Зная, когда все потоки завершены, и обрабатывая исключения

Question

Я использую среду Executor, чтобы запустить несколько потоков, используя пул потоков, т.е. newFixedThreadPool. Я использую threadpool.submit (aThread) для отправки заданий, которые будут выполняться пулом потоков, и это прекрасно работает, однако мне нужно определить, когда все потоки завершены, чтобы я мог продолжить другую обработку. Я посмотрел на использование Future.get (), который блокирует, пока поток не завершится, проблема в том, что он блокируется, пока не станет доступен результат. Я также рассмотрел использование непрерывного вызова метода isTeridity (), за которым следует спящий режим после выдачи завершения, чтобы проверить, все ли потоки завершены, но мне это не кажется опрятным. Есть ли другой способ чище? Также, если в каком-либо из потоков возникает исключение, я хочу иметь возможность завершить все остальные запущенные потоки, а также остановить запуск любого потока в очереди в пуле. Каков наилучший механизм для этого?

С нетерпением ждем ваших ответов

TIA

Answer 1

Использовать ExecutorService # shutdown () , а затем ExecutorService # awaitTermination ()

Например:

ExecutorService service = Executors.newCachedThreadPool();
service.submit(...);
service.submit(...);
service.shutdown();
service.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);

// All tasks have now finished

Что касается уведомления вас о сбое задачи с исключением. Вам нужно будет предоставить ThreadFactory для ExecutorService, который устанавливает «необработанный обработчик исключений» для каждого создаваемого потока. Этот обработчик исключений может затем завершить выполняющиеся задачи.

Answer 2

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

Вы можете написать простую оболочку, которая сделает это для любого Runnable.

Или, следуя этому примеру, вы можете расширить идею обернуть любой Callable.

class CallbackTask<T>
  implements Callable<T>
{

  private final Callable<? extends T> task;

  private final Callback<T> callback;

  CallbackTask(Callable<? extends T> task, Callback<T> callback)
  {
    this.task = task;
    this.callback = callback;
  }

  public T call()
    throws Exception
  {
    try {
      T result = task.call();
      callback.complete(result);
      return result;
    }
    catch (Exception ex) {
      callback.failed(ex);
      throw ex;
    }
  }

}