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

Question

В ходе выполнения моей программы запускается несколько потоков. Количество потоков варьируется в зависимости от пользовательских настроек, но все они выполняют один и тот же метод с разными переменными.

В некоторых ситуациях требуется очистка в середине выполнения, часть этого останавливает все потоки, хотя я не хочу, чтобы они немедленно останавливались, я просто устанавливаю переменную, для которой они проверяют, которая завершает их. Проблема в том, что до остановки потока может пройти до 1/2 секунды. Тем не менее, я должен быть уверен, что все потоки остановлены, прежде чем очистка может продолжаться. Очистка выполняется из другого потока, так что технически мне нужен этот поток, чтобы дождаться завершения других потоков.

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

Благодарю.

Answer 1

Просто присоединяйтесь к ним один за другим:

for (Thread thread : threads) {
  thread.join();
}

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

Answer 2

Если вы используете Java 1.5 или выше, вы можете попробовать CyclicBarrier. Вы можете передать операцию очистки в качестве параметра конструктора и просто вызвать barrier.await() на все темы, когда есть необходимость в очистке.

Answer 3

Вы видели Executor занятия в java.util.concurrent? Вы можете запустить ваши темы через ExecutorService, Он дает вам один объект, который вы можете использовать для отмены потоков или ожидания их завершения.

Answer 4

Определите служебный метод (или методы) самостоятельно:

public static waitFor(Collection<? extends Thread) c) throws InterruptedException {
    for(Thread t : c) t.join();
}

Или вы можете иметь массив

public static waitFor(Thread[] ts) throws InterruptedException {
    waitFor(Arrays.asList(ts));
}

В качестве альтернативы вы можете посмотреть на использование CyclicBarrier в java.util.concurrent библиотека для реализации произвольной точки рандеву между несколькими потоками.

Answer 5

Если вы управляете созданием потоков (отправкой в ​​ExecutorService), то, по-видимому, вы можете использовать ExecutorCompletionService см. ExecutorCompletionService? Зачем нужен один, если у нас есть invokeAll? для разных ответов есть.

Если вы не управляете созданием потоков, вот подход, который позволяет вам присоединяться к потокам "один за другим по мере их завершения" (и знать, какой из них заканчивается первым и т. Д.), Навеянным классом ruby ThreadWait. По сути, обновляя "наблюдающие потоки", которые предупреждают о завершении других потоков, вы можете узнать, когда "следующий" поток из многих завершается.

Вы бы использовали что-то вроде этого:

JoinThreads join = new JoinThreads(threads);
for(int i = 0; i < threads.size(); i++) {
  Thread justJoined = join.joinNextThread();
  System.out.println("Done with a thread, just joined=" + justJoined);
}

И источник:

public static class JoinThreads {
  java.util.concurrent.LinkedBlockingQueue<Thread> doneThreads = 
      new LinkedBlockingQueue<Thread>();

  public JoinThreads(List<Thread> threads) {
    for(Thread t : threads) {
      final Thread joinThis = t;
      new Thread(new Runnable() {
        @Override
        public void run() {
          try {
            joinThis.join();
            doneThreads.add(joinThis);
          }
          catch (InterruptedException e) {
            // "should" never get here, since we control this thread and don't call interrupt on it
          }
        }
      }).start();
    }

  }

  Thread joinNextThread() throws InterruptedException {
    return doneThreads.take();
  }
}

Приятной частью этого является то, что он работает с общими потоками Java, без изменений, любой поток может быть присоединен. Предостережение заключается в том, что требуется создание дополнительных потоков. Также эта конкретная реализация "оставляет потоки позади", если вы не вызываете joinNextThread() полное количество раз, и у вас нет метода "close" и т. Д. Прокомментируйте здесь, если вы хотите создать более совершенную версию. Вы также можете использовать этот же тип паттерна с "Futures" вместо объектов Thread и т. Д.