如何判断线程池任务执行完？

向右看齐 2023-10-14 13:59 13阅读 0赞

无论是在项目开发中，还是在面试中过程中，总会被问到或使用到并发编程来完成项目中的某个功能。

例如某个复杂的查询，无法使用一个查询语句来完成此功能，此时我们就需要执行多个查询语句，然后再将各自查询的结果，组装之后返回给前端了，那么这种场景下，我们就必须使用线程池来进行并发查询了。

> PS：磊哥做的最复杂的查询，总共关联了 21 张表，在和产品及需求方的沟通多次沟通下，才将查询的业务从 21 张表，降到了至少要查询 12 张表（非常难搞），那么这种场景下是无法使用一个查询语句来实现的，那么并发查询是必须要给安排上的。

# 1.需求分析 #

线程池的使用并不复杂，麻烦的是如何判断线程池中的任务已经全部执行完了？因为我们要等所有任务都执行完之后，才能进行数据的组装和返回，所以接下来，我们就来看如何判断线程中的任务是否已经全部执行完？

# 2.实现概述 #

判断线程池中的任务是否执行完的方法有很多，比如以下几个：

1.  使用 getCompletedTaskCount() 统计已经执行完的任务，和 getTaskCount() 线程池的总任务进行对比，如果相等则说明线程池的任务执行完了，否则既未执行完。
2.  使用 FutureTask 等待所有任务执行完，线程池的任务就执行完了。
3.  使用 CountDownLatch 或 CyclicBarrier 等待所有线程都执行完之后，再执行后续流程。

具体实现代码如下。

# 3.具体实现 #

# 3.1 统计完成任务数 #

通过判断线程池中的计划执行任务数和已完成任务数，来判断线程池是否已经全部执行完，如果**计划执行任务数=已完成任务数**，那么线程池的任务就全部执行完了，否则就未执行完。

示例代码如下：

java复制代码private static void isCompletedByTaskCount(ThreadPoolExecutor threadPool) {
        while (threadPool.getTaskCount() != threadPool.getCompletedTaskCount()) {
        }
    }

以上程序执行结果如下：

![format_png][]

![get_code_NzZhNmIyZWQyM2NmMWE5NGJlYjY2ZDFhMGI0ZWVmZmIsMTY5MTIwNDUwNDI0MQ][]

**方法说明**

*  getTaskCount()：返回计划执行的任务总数。由于任务和线程的状态可能在计算过程中动态变化，因此返回的值只是一个近似值。
 *  getCompletedTaskCount()：返回完成执行任务的总数。因为任务和线程的状态可能在计算过程中动态地改变，所以返回的值只是一个近似值，但是在连续的调用中并不会减少。

**缺点分析**

此判断方法的缺点是 getTaskCount() 和 getCompletedTaskCount() 返回的是一个近似值，因为线程池中的任务和线程的状态可能在计算过程中动态变化，所以它们两个返回的都是一个近似值。

# 3.2 FutureTask #

FutrueTask 的优势是任务判断精准，调用每个 FutrueTask 的 get 方法就是等待该任务执行完，如下代码所示：

java复制代码import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.FutureTask;
    
    /**
     * 使用 FutrueTask 等待线程池执行完全部任务
     */
    public class FutureTaskDemo {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            // 创建一个固定大小的线程池
            ExecutorService executor = Executors.newFixedThreadPool(3);
            // 创建任务
            FutureTask<Integer> task1 = new FutureTask<>(() -> {
                System.out.println("Task 1 start");
                Thread.sleep(2000);
                System.out.println("Task 1 end");
                return 1;
            });
            FutureTask<Integer> task2 = new FutureTask<>(() -> {
                System.out.println("Task 2 start");
                Thread.sleep(3000);
                System.out.println("Task 2 end");
                return 2;
            });
            FutureTask<Integer> task3 = new FutureTask<>(() -> {
                System.out.println("Task 3 start");
                Thread.sleep(1500);
                System.out.println("Task 3 end");
                return 3;
            });
            // 提交三个任务给线程池
            executor.submit(task1);
            executor.submit(task2);
            executor.submit(task3);
    
            // 等待所有任务执行完毕并获取结果
            int result1 = task1.get();
            int result2 = task2.get();
            int result3 = task3.get();
            System.out.println("Do main thread.");
        }
    }

以上程序的执行结果如下：

![image.png][]

![get_code_M2JkNGVkOGEyOGEzNjE3NzAwNjI4YWJkMTEzYmE4ODYsMTY5MTIwNDUwNDI0MQ][]

# 3.3 CountDownLatch和CyclicBarrier #

CountDownLatch 和 CyclicBarrier 类似，都是等待所有任务到达某个点之后，再进行后续的操作，如下图所示：

![d39a0ac7875c45669e42cd067334bb95.gif][]

![get_code_NjRmZjU4Y2RkNTcwMWU0NWFlOTIyZjU0NjE4ZDU0YTEsMTY5MTIwNDUwNDI0MQ][]

CountDownLatch 使用的示例代码如下：

java复制代码public static void main(String[] args) throws InterruptedException {
        // 创建线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
        	0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
        final int taskCount = 5;    // 任务总数
        // 单次计数器
        CountDownLatch countDownLatch = new CountDownLatch(taskCount); // ①
        // 添加任务
        for (int i = 0; i < taskCount; i++) {
            final int finalI = i;
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 随机休眠 0-4s
                        int sleepTime = new Random().nextInt(5);
                        TimeUnit.SECONDS.sleep(sleepTime);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(String.format("任务%d执行完成", finalI));
                    // 线程执行完，计数器 -1
                    countDownLatch.countDown();  // ②
                }
            });
        }
        // 阻塞等待线程池任务执行完
        countDownLatch.await();  // ③
        // 线程池执行完
        System.out.println();
        System.out.println("线程池任务执行完成！");
    }

> 代码说明：以上代码中标识为 ①、②、③ 的代码行是核心实现代码，其中：
> 
> ① 是声明一个包含了 5 个任务的计数器；
> 
> ② 是每个任务执行完之后计数器 -1；
> 
> ③ 是阻塞等待计数器 CountDownLatch 减为 0，表示任务都执行完了，可以执行 await 方法后面的业务代码了。

以上程序的执行结果如下：

![image.png][image.png 1]

![get_code_YzdkYjVkZGFhY2ZkZDlkNzgwODUwODk5ODZhZmVkYjAsMTY5MTIwNDUwNDI0MQ][]

**缺点分析**

CountDownLatch 缺点是计数器只能使用一次，CountDownLatch 创建之后不能被重复使用。

CyclicBarrier 和 CountDownLatch 类似，它可以理解为一个可以重复使用的循环计数器，CyclicBarrier 可以调用 reset 方法将自己重置到初始状态，CyclicBarrier 具体实现代码如下：

java复制代码public static void main(String[] args) throws InterruptedException {
        // 创建线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
        	0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
        final int taskCount = 5;    // 任务总数
        // 循环计数器 ①
        CyclicBarrier cyclicBarrier = new CyclicBarrier(taskCount, new Runnable() {
            @Override
            public void run() {
                // 线程池执行完
                System.out.println();
                System.out.println("线程池所有任务已执行完！");
            }
        });
        // 添加任务
        for (int i = 0; i < taskCount; i++) {
            final int finalI = i;
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 随机休眠 0-4s
                        int sleepTime = new Random().nextInt(5);
                        TimeUnit.SECONDS.sleep(sleepTime);
                        System.out.println(String.format("任务%d执行完成", finalI));
                        // 线程执行完
                        cyclicBarrier.await(); // ②
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

以上程序的执行结果如下：

![image.png][image.png 2]

![get_code_NDg4MDRmZjhiZjFkODBiMGNhODk0YzIyMDE4NzYwNTcsMTY5MTIwNDUwNDI0MQ][]

**方法说明**

CyclicBarrier 有 3 个重要的方法：

1.  构造方法：构造方法可以传递两个参数，参数 1 是计数器的数量 parties，参数 2 是计数器为 0 时，也就是任务都执行完之后可以执行的事件（方法）。
2.  await 方法：在 CyclicBarrier 上进行阻塞等待，当调用此方法时 CyclicBarrier 的内部计数器会 -1，直到发生以下情形之一： 在 CyclicBarrier 上等待的线程数量达到 parties，也就是计数器的声明数量时，则所有线程被释放，继续执行。 当前线程被中断，则抛出 InterruptedException 异常，并停止等待，继续执行。 其他等待的线程被中断，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。 其他等待的线程超时，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。 其他线程调用 CyclicBarrier.reset() 方法，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。
3.  reset 方法：使得CyclicBarrier回归初始状态，直观来看它做了两件事： 如果有正在等待的线程，则会抛出 BrokenBarrierException 异常，且这些线程停止等待，继续执行。 将是否破损标志位 broken 置为 false。

**优缺点分析**

CyclicBarrier 从设计的复杂度到使用的复杂度都高于 CountDownLatch，相比于 CountDownLatch 来说它的优点是可以重复使用（只需调用 reset 就能恢复到初始状态），缺点是使用难度较高。

# 小结 #

在实现判断线程池任务是否执行完成的方案中，通过统计线程池执行完任务的方式（实现方法 1），以及实现方法 3（CountDownLatch 或 CyclicBarrier）等统计，都是“不记名”的，只关注数量，不关注（具体）对象，所以这些方式都有可能受到外界代码的影响，因此使用 FutureTask 等待具体任务执行完的方式是最推荐的判断方法。

[format_png]: https://img-blog.csdnimg.cn/img_convert/bd0c68d991b4002cdc8e527c64d08e4e.webp?x-oss-process=image/format,png
[get_code_NzZhNmIyZWQyM2NmMWE5NGJlYjY2ZDFhMGI0ZWVmZmIsMTY5MTIwNDUwNDI0MQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=NzZhNmIyZWQyM2NmMWE5NGJlYjY2ZDFhMGI0ZWVmZmIsMTY5MTIwNDUwNDI0MQ==
[image.png]: https://img-blog.csdnimg.cn/img_convert/df09a0c8215fd64aeb65929fa5641ccd.webp?x-oss-process=image/format,png
[get_code_M2JkNGVkOGEyOGEzNjE3NzAwNjI4YWJkMTEzYmE4ODYsMTY5MTIwNDUwNDI0MQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=M2JkNGVkOGEyOGEzNjE3NzAwNjI4YWJkMTEzYmE4ODYsMTY5MTIwNDUwNDI0MQ==
[d39a0ac7875c45669e42cd067334bb95.gif]: https://img-blog.csdnimg.cn/d39a0ac7875c45669e42cd067334bb95.gif
[get_code_NjRmZjU4Y2RkNTcwMWU0NWFlOTIyZjU0NjE4ZDU0YTEsMTY5MTIwNDUwNDI0MQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=NjRmZjU4Y2RkNTcwMWU0NWFlOTIyZjU0NjE4ZDU0YTEsMTY5MTIwNDUwNDI0MQ==
[image.png 1]: https://img-blog.csdnimg.cn/img_convert/83a9f39905ae2a861067a7eae006b43a.webp?x-oss-process=image/format,png
[get_code_YzdkYjVkZGFhY2ZkZDlkNzgwODUwODk5ODZhZmVkYjAsMTY5MTIwNDUwNDI0MQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=YzdkYjVkZGFhY2ZkZDlkNzgwODUwODk5ODZhZmVkYjAsMTY5MTIwNDUwNDI0MQ==
[image.png 2]: https://img-blog.csdnimg.cn/img_convert/dfa19bbfbce03eafb467479243ea0614.webp?x-oss-process=image/format,png
[get_code_NDg4MDRmZjhiZjFkODBiMGNhODk0YzIyMDE4NzYwNTcsMTY5MTIwNDUwNDI0MQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=NDg4MDRmZjhiZjFkODBiMGNhODk0YzIyMDE4NzYwNTcsMTY5MTIwNDUwNDI0MQ==