重构并发流程控制工具：告别CountDownLatch，拥抱CompletableFuture

2023-09-01 13:24:01

前言

大家好，我是小郭，前段时间使用了CountDownLatch来做并发流程的控制，在生产上碰到了一些问题，最终问题是解决了，但是那篇文章的评论大家让我用CompletableFuture来试一试，这篇文章我们就来看看使用CompletableFuture来改造我们的并发流程控制工具。

CountDownLatch的使用

CountDownLatch是一个简单的并发控制工具，它允许一个或多个线程等待其他线程完成任务。CountDownLatch的构造函数需要指定一个计数器，该计数器表示需要等待的任务数量。当一个任务完成时，调用CountDownLatch的countDown()方法，计数器减1。当计数器减至0时，所有等待的线程都会被唤醒。

CountDownLatch的使用方法非常简单，只需要遵循以下步骤：

创建一个CountDownLatch对象，并指定需要等待的任务数量。
在每个任务中，调用CountDownLatch的countDown()方法，表示任务已完成。
在主线程中，调用CountDownLatch的await()方法，等待所有任务完成。

CountDownLatch的优缺点

CountDownLatch是一个简单易用的并发控制工具，它非常适合用于需要等待多个任务完成的情况。但是，CountDownLatch也有一些缺点：

CountDownLatch只能用于等待固定数量的任务完成，如果任务数量不确定，则无法使用CountDownLatch。
CountDownLatch不能取消等待，如果在等待过程中需要取消等待，则无法使用CountDownLatch。

CompletableFuture的使用

CompletableFuture是一个更加灵活的并发控制工具，它可以用于等待一个或多个任务完成。CompletableFuture的构造函数不需要指定计数器，它可以等待任意数量的任务完成。当一个任务完成时，调用CompletableFuture的complete()方法，表示任务已完成。当所有任务完成时，CompletableFuture的状态变为已完成，所有等待的线程都会被唤醒。

CompletableFuture的使用方法也非常简单，只需要遵循以下步骤：

创建一个CompletableFuture对象。
在每个任务中，调用CompletableFuture的complete()方法，表示任务已完成。
在主线程中，调用CompletableFuture的get()方法，等待所有任务完成。

CompletableFuture的优缺点

CompletableFuture是一个非常灵活的并发控制工具，它可以用于等待任意数量的任务完成。CompletableFuture还可以取消等待，如果在等待过程中需要取消等待，可以使用CompletableFuture的cancel()方法。

但是，CompletableFuture也有一些缺点：

CompletableFuture的API更加复杂，学习曲线更陡峭。
CompletableFuture的性能不如CountDownLatch，在某些情况下，使用CountDownLatch可能会获得更好的性能。

生产环境中遇到的问题

在生产环境中，我使用CountDownLatch来控制并发流程，但是遇到了以下问题：

任务数量不确定，无法使用CountDownLatch。
需要在等待过程中取消等待，无法使用CountDownLatch。

如何使用CompletableFuture来解决问题

为了解决生产环境中遇到的问题，我将CountDownLatch替换成了CompletableFuture。CompletableFuture可以等待任意数量的任务完成，并且可以取消等待。因此，使用CompletableFuture可以解决生产环境中遇到的问题。

将CountDownLatch替换成CompletableFuture后，代码如下：

import java.util.concurrent.CompletableFuture;

public class Main {

    public static void main(String[] args) {
        // 创建一个CompletableFuture对象
        CompletableFuture<Void> future = new CompletableFuture<>();

        // 创建一个任务列表
        List<Runnable> tasks = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            tasks.add(() -> {
                // 模拟任务执行
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                // 任务执行完成，调用CompletableFuture的complete()方法
                future.complete(null);
            });
        }

        // 提交任务到线程池执行
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        executorService.invokeAll(tasks);

        // 等待所有任务完成
        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        // 关闭线程池
        executorService.shutdown();
    }
}