JAVA CountDownLatch与CyclicBarrier使用差异及踩坑点对比分析 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Java并发编程：CountDownLatch与CyclicBarrier深度解析及应用避坑指南

大家好，今天我们来深入探讨Java并发编程中两个重要的同步工具类：CountDownLatch 和 CyclicBarrier。它们都用于协调多个线程的执行，但应用场景和实现机制存在显著差异。我们将通过对比分析，代码示例和常见问题分析，帮助大家更好地理解和运用这两个工具。

一、CountDownLatch：倒计时器

CountDownLatch 可以理解为一个倒计时器，它允许一个或多个线程等待其他线程完成操作。其核心机制是维护一个计数器，初始值大于等于1，每次一个线程完成任务，计数器减1，当计数器变为0时，所有等待的线程被释放。

1.1 工作原理

初始化: CountDownLatch 通过构造函数传入一个初始计数。
countDown() 方法: 每个完成任务的线程调用 countDown() 方法，计数器减1。
await() 方法: 一个或多个线程调用 await() 方法进入阻塞状态，直到计数器变为0。
一次性使用: CountDownLatch 的计数器一旦变为0，就不能重置，这意味着它是一次性使用的。

1.2 代码示例

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

public class CountDownLatchExample {

    public static void main(String[] args) throws InterruptedException {
        int numberOfThreads = 3;
        CountDownLatch latch = new CountDownLatch(numberOfThreads);

        ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            final int taskId = i;
            executor.execute(() -> {
                try {
                    System.out.println("Thread " + taskId + " is running...");
                    Thread.sleep((long) (Math.random() * 2000)); // 模拟任务执行时间
                    System.out.println("Thread " + taskId + " finished.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown(); // 任务完成，计数器减1
                }
            });
        }

        latch.await(); // 主线程等待所有子线程完成

        System.out.println("All threads finished. Main thread resuming.");
        executor.shutdown();
    }
}

1.3 适用场景

等待多个任务完成: 主线程需要等待多个子线程完成初始化任务后才能继续执行。
并行计算，汇总结果: 将一个大任务分解成多个小任务并行执行，最后等待所有任务完成，汇总结果。
测试框架: 在单元测试中，可以用来确保所有异步操作都已完成。

1.4 常见问题及避免

计数器初始值错误: 初始值必须等于需要等待的线程数量，否则可能导致 await() 方法永远阻塞或者提前释放。
线程未调用 countDown(): 如果某个线程由于异常或其他原因没有调用 countDown() 方法，计数器将无法减到0，导致 await() 方法永久阻塞。务必在 finally 块中调用 countDown() 方法，确保无论发生什么情况，计数器都会被递减。
线程中断: await() 方法会响应中断。如果线程在等待期间被中断，会抛出 InterruptedException。需要妥善处理中断异常，例如重新设置中断状态或者退出等待。

二、CyclicBarrier：循环栅栏

CyclicBarrier 允许一组线程互相等待，直到所有线程都到达一个公共点（栅栏），然后所有线程才能继续执行。与 CountDownLatch 不同的是，CyclicBarrier 可以重用，线程到达栅栏后可以再次启动新一轮的等待。

2.1 工作原理

初始化: CyclicBarrier 通过构造函数传入参与线程的数量和一个可选的 Runnable 任务，该任务会在所有线程到达栅栏后执行。
await() 方法: 每个线程调用 await() 方法进入等待状态，直到所有线程都到达栅栏。
栅栏打开: 当所有线程都到达栅栏后，栅栏被打开，所有线程被释放，可以继续执行。
Runnable 任务执行: 如果构造函数中提供了 Runnable 任务，则该任务会在栅栏打开前由最后一个到达的线程执行。
重用: 栅栏打开后，可以再次使用，参与线程可以再次调用 await() 方法进入新一轮的等待。

2.2 代码示例

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CyclicBarrierExample {

    public static void main(String[] args) {
        int numberOfThreads = 3;
        CyclicBarrier barrier = new CyclicBarrier(numberOfThreads, () -> {
            System.out.println("All threads have reached the barrier. Executing barrier action.");
        });

        ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            final int taskId = i;
            executor.execute(() -> {
                try {
                    System.out.println("Thread " + taskId + " is working...");
                    Thread.sleep((long) (Math.random() * 2000)); // 模拟任务执行时间
                    System.out.println("Thread " + taskId + " is waiting at the barrier.");
                    barrier.await(); // 等待其他线程到达栅栏
                    System.out.println("Thread " + taskId + " passed the barrier and is continuing.");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

2.3 适用场景

并行计算的迭代过程: 在迭代计算中，每个线程完成一次迭代后需要同步，然后才能开始下一次迭代。
多线程游戏: 在游戏中，多个玩家需要等待所有玩家都准备好后才能开始游戏。
数据分析 pipeline: 在数据分析中，多个步骤需要依赖前一个步骤的结果，可以使用 CyclicBarrier 来同步每个步骤的执行。

2.4 常见问题及避免

BrokenBarrierException: 如果在等待过程中，某个线程被中断，或者栅栏被重置，则其他等待的线程会抛出 BrokenBarrierException。需要捕获此异常，并根据业务逻辑进行处理，例如重试或者退出。
线程数量不匹配: 如果到达栅栏的线程数量少于构造函数中指定的数量，则所有线程都会一直等待，形成死锁。确保所有参与线程都正确地调用 await() 方法。
栅栏重置: CyclicBarrier 提供了 reset() 方法来重置栅栏。如果需要在等待过程中强制重置栅栏，需要小心处理，避免导致其他线程抛出 BrokenBarrierException。
死锁: 避免在栅栏动作中等待栅栏中的线程。这会很容易导致死锁，因为栅栏动作是由栅栏中的线程执行的，所以它永远不会完成。

三、CountDownLatch vs CyclicBarrier：对比分析

特性	CountDownLatch	CyclicBarrier
用途	一个或多个线程等待其他线程完成操作	一组线程互相等待，直到所有线程都到达栅栏
重用性	一次性使用，计数器无法重置	可以重用，栅栏可以循环使用
计数器机制	递减计数器，计数器变为0时释放等待线程	基于参与线程数量的计数器，所有线程到达时释放
异常处理	主要关注 `InterruptedException`	主要关注 `InterruptedException` 和 `BrokenBarrierException`
适用场景	等待多个任务完成，汇总结果	并行计算的迭代过程，多线程游戏
栅栏动作	无	可以指定一个 `Runnable` 任务在栅栏打开前执行
主要区别	CountDownLatch主要是让某些线程去等待另一些线程完成任务，CyclicBarrier主要是让一组线程相互等待至某个状态后，再全部同时执行。

选择建议:

如果只需要等待一组线程完成任务，且不需要重用同步机制，则 CountDownLatch 更简单直接。
如果需要一组线程在迭代计算中互相同步，并且可以重用同步机制，则 CyclicBarrier 更适合。

四、代码实例：模拟赛跑

我们可以使用 CountDownLatch 和 CyclicBarrier 来模拟赛跑场景，进一步理解它们的差异。

4.1 使用 CountDownLatch 模拟赛跑

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

public class RaceWithCountDownLatch {

    public static void main(String[] args) throws InterruptedException {
        int numberOfRunners = 5;
        CountDownLatch startSignal = new CountDownLatch(1); // 发令枪
        CountDownLatch finishSignal = new CountDownLatch(numberOfRunners); // 终点线

        ExecutorService executor = Executors.newFixedThreadPool(numberOfRunners);

        for (int i = 0; i < numberOfRunners; i++) {
            final int runnerId = i;
            executor.execute(() -> {
                try {
                    System.out.println("Runner " + runnerId + " is ready.");
                    startSignal.await(); // 等待发令枪响
                    System.out.println("Runner " + runnerId + " is running...");
                    Thread.sleep((long) (Math.random() * 5000)); // 模拟跑步时间
                    System.out.println("Runner " + runnerId + " finished.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    finishSignal.countDown(); // 到达终点
                }
            });
        }

        System.out.println("The race is about to start...");
        Thread.sleep(1000); // 模拟准备时间
        startSignal.countDown(); // 发令枪响

        finishSignal.await(); // 等待所有选手到达终点

        System.out.println("All runners finished. The race is over.");
        executor.shutdown();
    }
}

4.2 使用 CyclicBarrier 模拟赛跑

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RaceWithCyclicBarrier {

    public static void main(String[] args) {
        int numberOfRunners = 5;
        CyclicBarrier startBarrier = new CyclicBarrier(numberOfRunners, () -> {
            System.out.println("All runners are ready. The race starts!");
        });

        CyclicBarrier finishBarrier = new CyclicBarrier(numberOfRunners, () -> {
            System.out.println("All runners have finished this lap!");
        });

        ExecutorService executor = Executors.newFixedThreadPool(numberOfRunners);

        for (int i = 0; i < numberOfRunners; i++) {
            final int runnerId = i;
            executor.execute(() -> {
                try {
                    for (int lap = 1; lap <= 2; lap++) { // 跑两圈
                        System.out.println("Runner " + runnerId + " is ready for lap " + lap + ".");
                        startBarrier.await(); // 等待所有选手准备好
                        System.out.println("Runner " + runnerId + " is running lap " + lap + "...");
                        Thread.sleep((long) (Math.random() * 3000)); // 模拟跑步时间
                        System.out.println("Runner " + runnerId + " finished lap " + lap + ".");
                        finishBarrier.await(); // 等待所有选手完成此圈
                    }
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown();
    }
}

在这个 CyclicBarrier 的例子中，我们模拟了赛跑选手跑两圈的过程。每个 CyclicBarrier 的 await 方法代表着一次同步点。 startBarrier 和 finishBarrier 分别用于同步每一圈的开始和结束。

五、常见误用场景分析

将 CountDownLatch 用于循环同步： CountDownLatch 是一次性的，如果需要循环同步，每次循环都必须创建一个新的 CountDownLatch 实例，这会增加代码的复杂性，并且容易出错。 CyclicBarrier 才是循环同步的正确选择。
在 CyclicBarrier 的 barrierAction 中执行耗时操作： barrierAction 由到达栅栏的最后一个线程执行，如果 barrierAction 执行时间过长，会阻塞其他线程的执行，降低并发效率。应该尽量避免在 barrierAction 中执行耗时操作，或者将其异步执行。
忽略异常处理： 无论是 CountDownLatch 还是 CyclicBarrier，都可能抛出 InterruptedException 和 BrokenBarrierException 异常。忽略这些异常会导致程序行为不确定，甚至可能导致死锁。务必捕获这些异常，并进行妥善处理。
线程池大小设置不合理： 使用线程池时，如果线程池大小设置不合理，可能导致线程饥饿或者资源浪费。需要根据实际情况合理设置线程池大小。例如，如果任务是 CPU 密集型的，则线程池大小可以设置为 CPU 核心数 + 1。如果任务是 IO 密集型的，则线程池大小可以设置为 CPU 核心数的 2 倍甚至更多。

六、关键差异和选择要点回顾

CountDownLatch 是一次性的倒计时器，用于一个或多个线程等待其他线程完成任务。
CyclicBarrier 是可重用的栅栏，用于一组线程互相等待，直到所有线程都到达栅栏。
CountDownLatch 主要关注等待其他线程完成任务，CyclicBarrier 主要关注线程间的相互同步。

七、总结

掌握 CountDownLatch 和 CyclicBarrier 的使用，能够帮助我们编写更加高效和健壮的并发程序。理解它们的差异，选择合适的工具，并注意避免常见的陷阱，是提升并发编程能力的关键。