Java多种方法实现等待所有子线程完成再继续执行

简介

在现实世界中，我们常常需要等待其它任务完成，才能继续执行下一步。Java实现等待子线程完成再继续执行的方式很多。我们来一一查看一下。
Thread的join方法

该方法是Thread提供的方法，调用join()时，会阻塞主线程，等该Thread完成才会继续执行，代码如下：

1. private static void threadJoin() {
2.   List<Thread> threads = new ArrayList<>();
4.   for (int i = 0; i < NUM; i++) {
5.     Thread t = new Thread(new PkslowTask("Task " + i));
6.     t.start();
7.     threads.add(t);
8.   }
9.   threads.forEach(t -> {
10.     try {
11.       t.join();
12.     } catch (InterruptedException e) {
13.       throw new RuntimeException(e);
14.     }
15.   });
17.   System.out.println("threadJoin Finished All Tasks...");
19. }

复制代码

结果：

1. Task 6 is running
2. Task 9 is running
3. Task 3 is running
4. Task 4 is running
5. Task 7 is running
6. Task 0 is running
7. Task 2 is running
8. Task 1 is running
9. Task 5 is running
10. Task 8 is running
11. Task 1 is completed
12. Task 8 is completed
13. Task 6 is completed
14. Task 4 is completed
15. Task 3 is completed
16. Task 0 is completed
17. Task 7 is completed
18. Task 9 is completed
19. Task 2 is completed
20. Task 5 is completed
21. threadJoin Finished All Tasks...

复制代码

CountDownLatch

CountDownLatch是一个很好用的并发工具，初始化时要指定线程数，如10。在子线程调用countDown()时计数减1。直到为0时，await()方法才不会阻塞。代码如下：

1. private static void countDownLatch() {
2.   CountDownLatch latch = new CountDownLatch(NUM);
3.   for (int i = 0; i < NUM; i++) {
4.     Thread t = new Thread(() -> {
5.       System.out.println("countDownLatch running...");
6.       try {
7.         Thread.sleep(1000);
8.         System.out.println("countDownLatch Finished...");
9.         latch.countDown();
10.       } catch (InterruptedException e) {
11.         throw new RuntimeException(e);
12.       }
13.     });
14.     t.start();
15.   }
17.   try {
18.     latch.await();
19.   } catch (InterruptedException e) {
20.     throw new RuntimeException(e);
21.   }
22.   System.out.println("countDownLatch Finished All Tasks...");
23. }

复制代码

结果：

1. countDownLatch running...
2. countDownLatch running...
3. countDownLatch running...
4. countDownLatch running...
5. countDownLatch running...
6. countDownLatch running...
7. countDownLatch running...
8. countDownLatch running...
9. countDownLatch running...
10. countDownLatch running...
11. countDownLatch Finished...
12. countDownLatch Finished...
13. countDownLatch Finished...
14. countDownLatch Finished...
15. countDownLatch Finished...
16. countDownLatch Finished...
17. countDownLatch Finished...
18. countDownLatch Finished...
19. countDownLatch Finished...
20. countDownLatch Finished...
21. countDownLatch Finished All Tasks...

复制代码

CyclicBarrier

CyclicBarrier与CountDownLatch类似，但CyclicBarrier可重置，可重用。代码如下：

1. private static void cyclicBarrier() {
2.   CyclicBarrier barrier = new CyclicBarrier(NUM + 1);
4.   for (int i = 0; i < NUM; i++) {
5.     Thread t = new Thread(() -> {
6.       System.out.println("cyclicBarrier running...");
7.       try {
8.         Thread.sleep(1000);
9.         System.out.println("cyclicBarrier Finished...");
10.         barrier.await();
11.       } catch (InterruptedException | BrokenBarrierException e) {
12.         throw new RuntimeException(e);
13.       }
14.     });
15.     t.start();
16.   }
18.   try {
19.     barrier.await();
20.   } catch (InterruptedException | BrokenBarrierException e) {
21.     throw new RuntimeException(e);
22.   }
23.   System.out.println("cyclicBarrier Finished All Tasks...");
24. }

复制代码

结果：

1. cyclicBarrier running...
2. cyclicBarrier running...
3. cyclicBarrier running...
4. cyclicBarrier running...
5. cyclicBarrier running...
6. cyclicBarrier running...
7. cyclicBarrier running...
8. cyclicBarrier running...
9. cyclicBarrier running...
10. cyclicBarrier running...
11. cyclicBarrier Finished...
12. cyclicBarrier Finished...
13. cyclicBarrier Finished...
14. cyclicBarrier Finished...
15. cyclicBarrier Finished...
16. cyclicBarrier Finished...
17. cyclicBarrier Finished...
18. cyclicBarrier Finished...
19. cyclicBarrier Finished...
20. cyclicBarrier Finished...
21. cyclicBarrier Finished All Tasks...

复制代码

executorService.isTerminated()

ExecutorService调用shutdown()方法后，可以通过方法isTerminated()来判断任务是否完成。代码如下：

1. private static void executeServiceIsTerminated() {
2.   ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
3.   IntStream.range(0, NUM)
4.     .forEach(i -> executorService.execute(new PkslowTask("Task " + i)));
5.   executorService.shutdown();
6.   while (!executorService.isTerminated()) {
7.     //waiting...
8.   }
9.   System.out.println("executeServiceIsTerminated Finished All Tasks...");
11. }

复制代码

结果：

1. Task 0 is running
2. Task 2 is running
3. Task 1 is running
4. Task 3 is running
5. Task 4 is running
6. Task 0 is completed
7. Task 2 is completed
8. Task 5 is running
9. Task 4 is completed
10. Task 7 is running
11. Task 3 is completed
12. Task 1 is completed
13. Task 8 is running
14. Task 6 is running
15. Task 9 is running
16. Task 5 is completed
17. Task 9 is completed
18. Task 7 is completed
19. Task 6 is completed
20. Task 8 is completed
21. executeServiceIsTerminated Finished All Tasks...

复制代码

executorService.awaitTermination

executorService.awaitTermination方法会等待任务完成，并给一个超时时间，代码如下：

1. private static void executeServiceAwaitTermination() {
2.   ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
3.   IntStream.range(0, NUM)
4.     .forEach(i -> executorService.execute(new PkslowTask("Task " + i)));
5.   executorService.shutdown();
7.   try {
8.     if (!executorService.awaitTermination(1, TimeUnit.MINUTES)) {
9.       executorService.shutdownNow();
10.     }
11.   } catch (InterruptedException e) {
12.     throw new RuntimeException(e);
13.   }
14.   System.out.println("executeServiceAwaitTermination Finished All Tasks...");
15. }

复制代码

结果：

1. Task 0 is running
2. Task 1 is running
3. Task 2 is running
4. Task 3 is running
5. Task 4 is running
6. Task 0 is completed
7. Task 5 is running
8. Task 1 is completed
9. Task 4 is completed
10. Task 7 is running
11. Task 3 is completed
12. Task 8 is running
13. Task 2 is completed
14. Task 9 is running
15. Task 6 is running
16. Task 5 is completed
17. Task 7 is completed
18. Task 9 is completed
19. Task 8 is completed
20. Task 6 is completed
21. executeServiceAwaitTermination Finished All Tasks...

复制代码

executorService.invokeAll

使用invokeAll提交所有任务，代码如下：

1. private static void executeServiceInvokeAll() {
2.   ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
3.   List<Callable<Void>> tasks = new ArrayList<>();
5.   IntStream.range(0, NUM)
6.     .forEach(i -> tasks.add(new PkslowTask("Task " + i)));
8.   try {
9.     executorService.invokeAll(tasks);
10.   } catch (InterruptedException e) {
11.     throw new RuntimeException(e);
12.   }
14.   executorService.shutdown();
15.   System.out.println("executeServiceInvokeAll Finished All Tasks...");
16. }

复制代码

结果：

1. Task 1 is running
2. Task 2 is running
3. Task 0 is running
4. Task 3 is running
5. Task 4 is running
6. Task 1 is completed
7. Task 3 is completed
8. Task 0 is completed
9. Task 2 is completed
10. Task 4 is completed
11. Task 8 is running
12. Task 5 is running
13. Task 6 is running
14. Task 9 is running
15. Task 7 is running
16. Task 8 is completed
17. Task 5 is completed
18. Task 6 is completed
19. Task 9 is completed
20. Task 7 is completed
21. executeServiceInvokeAll Finished All Tasks...

复制代码

ExecutorCompletionService

ExecutorCompletionService通过take()方法，会返回最早完成的任务，代码如下：

1. private static void executorCompletionService() {
2.   ExecutorService executorService = Executors.newFixedThreadPool(10);
3.   CompletionService<String> service = new ExecutorCompletionService<>(executorService);
5.   List<Callable<String>> callables = new ArrayList<>();
6.   callables.add(new DelayedCallable(2000, "2000ms"));
7.   callables.add(new DelayedCallable(1500, "1500ms"));
8.   callables.add(new DelayedCallable(6000, "6000ms"));
9.   callables.add(new DelayedCallable(2500, "2500ms"));
10.   callables.add(new DelayedCallable(300, "300ms"));
11.   callables.add(new DelayedCallable(3000, "3000ms"));
12.   callables.add(new DelayedCallable(1100, "1100ms"));
13.   callables.add(new DelayedCallable(100, "100ms"));
14.   callables.add(new DelayedCallable(100, "100ms"));
15.   callables.add(new DelayedCallable(100, "100ms"));
17.   callables.forEach(service::submit);
19.   for (int i = 0; i < NUM; i++) {
20.     try {
21.       Future<String> future = service.take();
22.       System.out.println(future.get() + " task is completed");
23.     } catch (InterruptedException | ExecutionException e) {
24.       throw new RuntimeException(e);
25.     }
26.   }
28.   System.out.println("executorCompletionService Finished All Tasks...");
30.   executorService.shutdown();
31.   awaitTerminationAfterShutdown(executorService);
32. }

复制代码

这里不同任务的时长是不一样的，但会先返回最早完成的任务：

1. 2000ms is running
2. 2500ms is running
3. 300ms is running
4. 1500ms is running
5. 6000ms is running
6. 3000ms is running
7. 1100ms is running
8. 100ms is running
9. 100ms is running
10. 100ms is running
11. 100ms is completed
12. 100ms is completed
13. 100ms task is completed
14. 100ms task is completed
15. 100ms is completed
16. 100ms task is completed
17. 300ms is completed
18. 300ms task is completed
19. 1100ms is completed
20. 1100ms task is completed
21. 1500ms is completed
22. 1500ms task is completed
23. 2000ms is completed
24. 2000ms task is completed
25. 2500ms is completed
26. 2500ms task is completed
27. 3000ms is completed
28. 3000ms task is completed
29. 6000ms is completed
30. 6000ms task is completed
31. executorCompletionService Finished All Tasks...

复制代码

代码

代码请看GitHub: https://github.com/LarryDpk/pkslow-samples

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！