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

Ответ 1

В основном на ExecutorService вы вызываете shutdown(), а затем awaitTermination():

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}
taskExecutor.shutdown();
try {
  taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
  ...
}

Ответ 2

Используйте CountDownLatch:

CountDownLatch latch = new CountDownLatch(totalNumberOfTasks);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}

try {
  latch.await();
} catch (InterruptedException E) {
   // handle
}

и внутри вашей задачи (заключить в try/finally)

latch.countDown();

Ответ 3

ExecutorService.invokeAll() делает это для вас.

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
List<Callable<?>> tasks; // your tasks
// invokeAll() returns when all tasks are complete
List<Future<?>> futures = taskExecutor.invokeAll(tasks);

Ответ 4

Вы также можете использовать Списки фьючерсов:

List<Future> futures = new ArrayList<Future>();
// now add to it:
futures.add(executorInstance.submit(new Callable<Void>() {
  public Void call() throws IOException {
     // do something
    return null;
  }
}));

тогда, когда вы хотите присоединиться ко всем из них, это, по сути, эквивалент объединения каждого из них (с добавленной выгодой, что он повторно вызывает исключения из дочерних потоков в основной):

for(Future f: this.futures) { f.get(); }

В принципе, трюк заключается в вызове .get() для каждого Будущего один за раз, вместо бесконечного цикла, вызывающего isDone() on (все или каждый). Таким образом, вы сможете "перейти" через этот блок и пройти мимо него, как только закончится последний поток. Предостережение заключается в том, что, поскольку вызов .get() повторно вызывает исключения, если один из потоков умирает, вы могли бы увеличить его от этого, возможно, до того, как другие потоки завершили выполнение [во избежание этого вы можете добавить catch ExecutionException вокруг вызов get]. Другое предостережение - это ссылка на все потоки, поэтому, если у них есть локальные переменные потока, они не будут собираться до тех пор, пока вы не пройдете мимо этого блока (хотя вы могли бы обойти это, если бы это стало проблемой, путем удаления Будущее от ArrayList). Если вы хотите знать, какое будущее "заканчивается первым", вы можете использовать что-то вроде fooobar.com/questions/25029/...

Ответ 5

В Java8 вы можете сделать это с помощью CompletableFuture:

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();

Ответ 6

Просто мои два цента. Чтобы преодолеть требование CountDownLatch, чтобы узнать количество заданий заранее, вы можете сделать это по-старому, используя простой Semaphore.

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
int numberOfTasks=0;
Semaphore s=new Semaphore(0);
while(...) {
    taskExecutor.execute(new MyTask());
    numberOfTasks++;
}

try {
    s.aquire(numberOfTasks);
...

В своей задаче просто вызовите s.release(), как вы бы latch.countDown();

Ответ 7

Класс CyclicBarrier в Java 5 и более поздних версиях предназначен для такого рода вещей.

Ответ 8

Немного поздно в игре, но ради завершения...

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

Guava предлагает несколько интересных инструментов для этого.

Пример::

Оберните службу ExecutorService в ListeningExecutorService::

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));

Отправить коллекцию вызовов для выполнения::

for (Callable<Integer> callable : callables) {
  ListenableFuture<Integer> lf = service.submit(callable);
  // listenableFutures is a collection
  listenableFutures.add(lf)
});

Теперь важная часть:

ListenableFuture<List<Integer>> lf = Futures.successfulAsList(listenableFutures);

Прикрепите обратный вызов к ListenableFuture, который вы можете использовать для уведомления, когда все фьючерсы завершены::

        Futures.addCallback(lf, new FutureCallback<List<Integer>>() {
        @Override
        public void onSuccess(List<Integer> result) {
            log.info("@@ finished processing {} elements", Iterables.size(result));
            // do something with all the results
        }

        @Override
        public void onFailure(Throwable t) {
            log.info("@@ failed because of :: {}", t);
        }
    });

Это также дает преимущество, что вы можете собрать все результаты в одном месте после завершения обработки...

Подробнее здесь

Ответ 9

Следуйте одному из следующих подходов.

1. Выполните итерацию всех будущих задач, возвращаемых из submit в ExecutorService, и проверяйте состояние с помощью блокирующего вызова get() для объекта Future, как предложено Kiran
2. Используйте invokeAll() на ExecutorService
3. CountDownLatch
4. ForkJoinPool или Executors.html # newWorkStealingPool
5. Используйте API-интерфейсы shutdown, awaitTermination, shutdownNow ThreadPoolExecutor в правильной последовательности

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

Как CountDownLatch используется в многопоточности Java?

Как правильно отключить Java ExecutorService

Ответ 10

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

taskExecutor.execute(new Runnable() {
  public void run() {
    taskStartedNotification();
    new MyTask().run();
    taskFinishedNotification();
  }
});

Ответ 11

Здесь есть два варианта, просто запутайте, какой из них лучше всего выбрать.

Опция 1:

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();

Вариант 2:

ExecutorService es = Executors.newFixedThreadPool(4);
List< Future<?>> futures = new ArrayList<>();
for(Runnable task : taskList) {
    futures.add(es.submit(task));
}

for(Future<?> future : futures) {
    try {
        future.get();
    }catch(Exception e){
        // do logging and nothing else
    }
}
es.shutdown();

Здесь положить future.get(); в попытке поймать это хорошая идея, верно?

Ответ 12

Я только что написал пример программы, которая решает вашу проблему. Данной краткой реализации не было, поэтому я добавлю ее. Хотя вы можете использовать executor.shutdown() и executor.awaitTermination(), это не лучшая практика, поскольку время, затрачиваемое разными потоками, было бы непредсказуемым.

ExecutorService es = Executors.newCachedThreadPool();
    List<Callable<Integer>> tasks = new ArrayList<>();

    for (int j = 1; j <= 10; j++) {
        tasks.add(new Callable<Integer>() {

            @Override
            public Integer call() throws Exception {
                int sum = 0;
                System.out.println("Starting Thread "
                        + Thread.currentThread().getId());

                for (int i = 0; i < 1000000; i++) {
                    sum += i;
                }

                System.out.println("Stopping Thread "
                        + Thread.currentThread().getId());
                return sum;
            }

        });
    }

    try {
        List<Future<Integer>> futures = es.invokeAll(tasks);
        int flag = 0;

        for (Future<Integer> f : futures) {
            Integer res = f.get();
            System.out.println("Sum: " + res);
            if (!f.isDone()) 
                flag = 1;
        }

        if (flag == 0)
            System.out.println("SUCCESS");
        else
            System.out.println("FAILED");

    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }

Ответ 13

Просто чтобы предоставить больше альтернатив здесь, чтобы использовать защелки/барьеры. Вы также можете получить частичные результаты, пока все они не закончили использование CompletionService.

Из Java Concurrency на практике: "Если у вас есть партия вычислений для представления Исполнителю, и вы хотите получить их результаты по мере их появления доступный, вы могли бы сохранить будущее, связанное с каждой задачей, и повторно опросить для завершения, позвонив время ожидания от нуля. Это возможно, но утомительно. К счастью, существует лучший способ: служба завершения. "

Здесь реализация

public class TaskSubmiter {
    private final ExecutorService executor;
    TaskSubmiter(ExecutorService executor) { this.executor = executor; }
    void doSomethingLarge(AnySourceClass source) {
        final List<InterestedResult> info = doPartialAsyncProcess(source);
        CompletionService<PartialResult> completionService = new ExecutorCompletionService<PartialResult>(executor);
        for (final InterestedResult interestedResultItem : info)
            completionService.submit(new Callable<PartialResult>() {
                public PartialResult call() {
                    return InterestedResult.doAnOperationToGetPartialResult();
                }
        });

    try {
        for (int t = 0, n = info.size(); t < n; t++) {
            Future<PartialResult> f = completionService.take();
            PartialResult PartialResult = f.get();
            processThisSegment(PartialResult);
            }
        } 
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } 
        catch (ExecutionException e) {
            throw somethinghrowable(e.getCause());
        }
    }
}

Ответ 14

Вы можете использовать этот код:

public class MyTask implements Runnable {

    private CountDownLatch countDownLatch;

    public MyTask(CountDownLatch countDownLatch {
         this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
         try {
             //Do somethings
             //
             this.countDownLatch.countDown();//important
         } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
         }
     }
}

CountDownLatch countDownLatch = new CountDownLatch(NUMBER_OF_TASKS);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
for (int i = 0; i < NUMBER_OF_TASKS; i++){
     taskExecutor.execute(new MyTask(countDownLatch));
}
countDownLatch.await();
System.out.println("Finish tasks");

Ответ 15

Это моё решение, основанное на подсказке AdamSkywalker, и оно работает

package frss.main;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestHilos {

    void procesar() {
        ExecutorService es = Executors.newFixedThreadPool(4);
        List<Runnable> tasks = getTasks();
        CompletableFuture<?>[] futures = tasks.stream().map(task -> CompletableFuture.runAsync(task, es)).toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(futures).join();
        es.shutdown();

        System.out.println("FIN DEL PROCESO DE HILOS");
    }

    private List<Runnable> getTasks() {
        List<Runnable> tasks = new ArrayList<Runnable>();

        Hilo01 task1 = new Hilo01();
        tasks.add(task1);

        Hilo02 task2 = new Hilo02();
        tasks.add(task2);
        return tasks;
    }

    private class Hilo01 extends Thread {

        @Override
        public void run() {
            System.out.println("HILO 1");
        }

    }

    private class Hilo02 extends Thread {

        @Override
        public void run() {
            try {
                sleep(2000);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("HILO 2");
        }

    }


    public static void main(String[] args) {
        TestHilos test = new TestHilos();
        test.procesar();
    }
}

Ответ 16

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

import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** Testing CountDownLatch and ExecutorService to manage scenario where
 * multiple Threads work together to complete tasks from a single
 * resource provider, so the processing can be faster. */
public class ThreadCountDown {

private CountDownLatch threadsCountdown = null;
private static Queue<Integer> tasks = new PriorityQueue<>();

public static void main(String[] args) {
    // Create a queue with "Tasks"
    int numberOfTasks = 2000;
    while(numberOfTasks-- > 0) {
        tasks.add(numberOfTasks);
    }

    // Initiate Processing of Tasks
    ThreadCountDown main = new ThreadCountDown();
    main.process(tasks);
}

/* Receiving the Tasks to process, and creating multiple Threads
* to process in parallel. */
private void process(Queue<Integer> tasks) {
    int numberOfThreads = getNumberOfThreadsRequired(tasks.size());
    threadsCountdown = new CountDownLatch(numberOfThreads);
    ExecutorService threadExecutor = Executors.newFixedThreadPool(numberOfThreads);

    //Initialize each Thread
    while(numberOfThreads-- > 0) {
        System.out.println("Initializing Thread: "+numberOfThreads);
        threadExecutor.execute(new MyThread("Thread "+numberOfThreads));
    }

    try {
        //Shutdown the Executor, so it cannot receive more Threads.
        threadExecutor.shutdown();
        threadsCountdown.await();
        System.out.println("ALL THREADS COMPLETED!");
        //continue With Some Other Process Here
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
}

/* Determine the number of Threads to create */
private int getNumberOfThreadsRequired(int size) {
    int threshold = 100;
    int threads = size / threshold;
    if( size > (threads*threshold) ){
        threads++;
    }
    return threads;
}

/* Task Provider. All Threads will get their task from here */
private synchronized static Integer getTask(){
    return tasks.poll();
}

/* The Threads will get Tasks and process them, while still available.
* When no more tasks available, the thread will complete and reduce the threadsCountdown */
private class MyThread implements Runnable {

    private String threadName;

    protected MyThread(String threadName) {
        super();
        this.threadName = threadName;
    }

    @Override
    public void run() {
        Integer task;
        try{
            //Check in the Task pool if anything pending to process
            while( (task = getTask()) != null ){
                processTask(task);
            }
        }catch (Exception ex){
            ex.printStackTrace();
        }finally {
            /*Reduce count when no more tasks to process. Eventually all
            Threads will end-up here, reducing the count to 0, allowing
            the flow to continue after threadsCountdown.await(); */
            threadsCountdown.countDown();
        }
    }

    private void processTask(Integer task){
        try{
            System.out.println(this.threadName+" is Working on Task: "+ task);
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }
}
}

Надеюсь, поможет!

Ответ 17

Вы можете использовать свой собственный подкласс ExecutorCompletionService для переноса taskExecutor, а ваша собственная реализация BlockingQueue, чтобы получить информацию, когда каждая задача завершается, и выполнять любые обратные вызовы или другие действия, которые вы желаете, когда количество завершенных задач достигает желаемой цели.

Ответ 18

вы должны использовать метод executorService.shutdown() и executorService.awaitTermination.

Пример следующим образом:

public class ScheduledThreadPoolExample {

    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
        executorService.scheduleAtFixedRate(() -> System.out.println("process task."),
                0, 1, TimeUnit.SECONDS);

        TimeUnit.SECONDS.sleep(10);
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.DAYS);
    }

}

Ответ 19

Поэтому я публикую свой ответ по связанному вопросу здесь, если кто-то хочет более простой способ сделать это

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture[] futures = new CompletableFuture[10];
int i = 0;
while (...) {
    futures[i++] =  CompletableFuture.runAsync(runner, executor);
}

CompletableFuture.allOf(futures).join(); // THis will wait until all future ready.

Ответ 20

Java 8 - Мы можем использовать stream API для обработки потока. См. Фрагмент ниже

final List<Runnable> tasks = ...; //or any other functional interface
tasks.stream().parallel().forEach(Runnable::run) // Uses default pool

//alternatively to specify parallelism 
new ForkJoinPool(15).submit(
          () -> tasks.stream().parallel().forEach(Runnable::run) 
    ).get();

Ответ 21

ExecutorService WORKER_THREAD_POOL 
  = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(2);
for (int i = 0; i < 2; i++) {
    WORKER_THREAD_POOL.submit(() -> {
        try {
            // doSomething();
            latch.countDown();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
}

// wait for the latch to be decremented by the two remaining threads
latch.await();

Если doSomething() выдает другие исключения, latch.countDown(), похоже, не будет выполняться, так что мне делать?

Ответ 22

если вы используете несколько потоков ExecutionServices ПОСЛЕДОВАТЕЛЬНО и хотите дождаться завершения КАЖДОГО ИСПОЛНИТЕЛЬНОГО ОБСЛУЖИВАНИЯ. Лучший способ, как показано ниже;

ExecutorService executer1 = Executors.newFixedThreadPool(THREAD_SIZE1);
for (<loop>) {
   executer1.execute(new Runnable() {
            @Override
            public void run() {
                ...
            }
        });
} 
executer1.shutdown();

try{
   executer1.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

   ExecutorService executer2 = Executors.newFixedThreadPool(THREAD_SIZE2);
   for (true) {
      executer2.execute(new Runnable() {
            @Override
            public void run() {
                 ...
            }
        });
   } 
   executer2.shutdown();
} catch (Exception e){
 ...
}

Ответ 23

Это может помочь

Log.i(LOG_TAG, "shutting down executor...");
executor.shutdown();
while (true) {
                try {
                    Log.i(LOG_TAG, "Waiting for executor to terminate...");
                    if (executor.isTerminated())
                        break;
                    if (executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                        break;
                    }
                } catch (InterruptedException ignored) {}
            }

Ответ 24

Вы можете вызвать waitTillDone() в этом классе бегуна :

Runner runner = Runner.runner(4); // create pool with 4 threads in thread pool

while(...) {
    runner.run(new MyTask()); // here you submit your task
}


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


runner.shutdown(); // once you done you can shutdown the runner

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

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

Более подробную информацию можно найти здесь:

https://github.com/MatejTymes/JavaFixes

Ответ 25

В исполнителе getActiveCount() есть метод, который дает количество активных потоков.

После прокрутки потока мы можем проверить, соответствует ли значение activeCount() 0. Когда значение равно нулю, подразумевается, что в настоящий момент нет активных потоков, что означает, что задача завершена:

while (true) {
    if (executor.getActiveCount() == 0) {
    //ur own piece of code
    break;
    }
}