Java 中的并发工具类详解：Semaphore、CountDownLatch 和 CyclicBarrier

向右看齐 2024-04-06 12:38 88阅读 0赞

**目录**

1、信号量：Semaphore

2、线程同步：CountDownLatch 和 CyclicBarrier

--------------------

**Java 并发包提供了哪些并发工具类？**

我们通常所说的并发包也就是 java.util.concurrent 及其子包，集中了 Java 并发的各种基础工具类，具体主要包括几个方面：

1.  提供了比 synchronized 更加高级的各种同步结构，包括 **CountDownLatch、CyclicBarrier、Semaphore 等**，可以实现更加丰富的多线程操作，比如利用 Semaphore 作为资源控制器，限制同时进行工作的线程数量。
2.  各种线程安全的容器，比如最常见的 ConcurrentHashMap、有序的 ConcurrentSkipListMap，或者通过类似快照机制，实现线程安全的动态数组 CopyOnWriteArrayList 等。
3.  各种并发队列实现，如各种 BlockingQueue 实现，比较典型的 ArrayBlockingQueue、 SynchronousQueue 或针对特定场景的 PriorityBlockingQueue 等。
4.  强大的 Executor 框架，可以创建各种不同类型的线程池，调度任务运行等，绝大部分情况下，不再需要自己从头实现线程池和任务调度器。

首先，我们来看看并发包提供的丰富同步结构：

*  CountDownLatch，允许一个或多个线程等待某些操作完成。// 同步等待
 *  CyclicBarrier，一种辅助性的同步结构，允许多个线程等待到达某个屏障。// 同步等待
 *  Semaphore，Java 版本的信号量实现。// 信号量

#### 1、信号量：Semaphore ####

**Java 提供了经典信号量（Semaphore）的实现，它通过控制一定数量的允许（permit）的方式，来达到限制通用资源访问的目的。**你可以想象一下这个场景，在车站、机场等出租车时，当很多空出租车就位时，为防止过度拥挤，调度员指挥排队等待坐车的队伍一次进来 5 个人上车，等这 5 个人坐车出发，再放进去下一批，这和 Semaphore 的工作原理有些类似。

试试使用 Semaphore 来模拟实现这个调度过程：

import java.util.concurrent.Semaphore;
    
    /**
     * 信号量：获取到信号量的程序，允许继续向下执行，比如电影院的固定座位数量
     */
    public class UsualSemaphoreSample {
    
        public static void main(String[] args) throws InterruptedException {
            System.out.println("Action...GO!");
            // 获取到信号量的程序，允许继续向下执行
            Semaphore semaphore = new Semaphore(5); // 初始化5个信号量
            for (int i = 0; i < 10; i++) {
                Thread t = new Thread(new SemaphoreWorker(semaphore));
                t.start();
            }
        }
    }
    
    class SemaphoreWorker implements Runnable {
    
        private String    name;
        private Semaphore semaphore;
    
        public SemaphoreWorker(Semaphore semaphore) {
            this.semaphore = semaphore;
        }
    
        @Override
        public void run() {
            try {
                log("尝试获取一个信号量!");
                semaphore.acquire(); // 获取一个信号量,如果获取不到信号量，将在这里被阻塞
                log("获取到信号量，继续执行...!");
                Thread.sleep(1000); // 模拟程序执行时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                log("释放一个信号量!");
                semaphore.release(); // 释放一个信号量
            }
        }
    
        private void log(String msg) {
            if (name == null) {
                name = Thread.currentThread().getName();
            }
            System.out.println(name + " " + msg);
        }
    }

这段代码是比较典型的 Semaphore 示例，其逻辑是，线程试图获得工作允许，得到许可则进行任务，然后释放许可，这时等待许可的其他线程，就可获得许可进入工作状态，直到全部处理结束。

编译运行，我们就能看到 Semaphore 的允许机制对工作线程的限制。

![ca7fd46a90234806a1047738d627d9e5.png][]

但是，从具体节奏来看，其实并不符合我们前面场景的需求，因为本例中 Semaphore 的用法实际是保证，一直有 5 个人可以试图乘车，如果有 1 个人出发了，立即就有排队的人获得许可，而这并不完全符合我们前面的要求。

那么，我再修改一下，演示个非典型的 Semaphore 用法。

import java.util.concurrent.Semaphore;
    
    /**
     * 信号量：获取到信号量的程序，允许继续向下执行，比如电影院的固定座位数量
     */
    public class AbnormalSemaphoreSample {
    
        public static void main(String[] args) throws InterruptedException {
            Semaphore semaphore = new Semaphore(0);    // 初始化信号量为 0
            for (int i = 0; i < 10; i++) {             // 准备好使用场景
                Thread t = new Thread(new MyWorker(semaphore));
                t.start();
            }
            System.out.println("Action...GO!");
            semaphore.release(5);                      // 释放 5 个信号量
            System.out.println("等待信号量用尽...");
            while (semaphore.availablePermits() != 0) {
                Thread.sleep(1000L);                   // 信号量没有用尽，一直循环阻塞
            }
            System.out.println("Action...GO again!");
            semaphore.release(5);
        }
    }
    
    class MyWorker implements Runnable {
    
        private Semaphore semaphore;
    
        public MyWorker(Semaphore semaphore) {
            this.semaphore = semaphore;
        }
    
        @Override
        public void run() {
            try {
                semaphore.acquire();  // 获取一个信号量
                System.out.println("Executed!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

![4d146379267145a885877034cfd6722a.png][]

注意，上面的代码，更侧重的是演示 Semaphore 的功能以及局限性，其实有很多线程编程中的反实践，比如使用了 sleep() 来协调任务执行，而且使用轮询调用 availalePermits() 来检测信号量获取情况，这都是很低效并且脆弱的，通常只是用在测试或者诊断场景。

总的来说，我们可以看出 **Semaphore 就是个计数器，其基本逻辑基于 acquire/release**，并没有太复杂的同步逻辑。

如果 Semaphore 的数值被初始化为 1，那么一个线程就可以通过 acquire 进入互斥状态，本质上和互斥锁是非常相似的。但是区别也非常明显，比如互斥锁是有持有者的，而对于 Semaphore 这种计数器结构，虽然有类似功能，但其实不存在真正意义的持有者，除非我们进行扩展包装。

#### 2、线程同步：CountDownLatch 和 CyclicBarrier ####

下面，来看看 CountDownLatch 和 CyclicBarrier，它们的行为有一定的相似度，经常会被考察二者有什么区别，简单总结一下：

1.  CountDownLatch 是不可以重置的，所以无法重用；而 CyclicBarrier 则没有这种限制，可以重用。
2.  CountDownLatch 的基本操作组合是 countDown/await。调用 await 的线程阻塞等待 countDown 足够的次数，不管你是在一个线程还是多个线程里 countDown，只要次数足够即可。所以就像 Brain Goetz 说过的，CountDownLatch 操作的是事件。// 可以在一个线程中多次的 countDown
3.  CyclicBarrier 的基本操作组合，则就是 await，当所有的伙伴（parties）都调用了 await，才会继续进行任务，并自动进行重置。注意，正常情况下，CyclicBarrier 的重置都是自动发生的，如果我们调用 reset 方法，但还有线程在等待，就会导致等待线程被打扰，抛出 BrokenBarrierException 异常。CyclicBarrier 侧重点是线程，而不是调用事件，它的典型应用场景是用来等待并发线程结束。// 一个线程只能 await 一次

如果用 CountDownLatch 去实现上面的排队场景，该怎么做呢？假设有 10 个人排队，我们将其分成 5 个人一批，通过 CountDownLatch 来协调批次，可以试试下面的示例代码。

import java.util.concurrent.CountDownLatch;
    
    /**
     * 同步阻塞
     */
    public class LatchSample {
    
        public static void main(String[] args) throws InterruptedException {
            // 等待 countDown 足够的次数，然后一起执行
            CountDownLatch latch = new CountDownLatch(6); // 不可重置，无法重用
            for (int i = 0; i < 5; i++) { // 第一批线程
                Thread t = new Thread(new FirstBatchWorker(latch));
                t.start();
            }
            for (int i = 0; i < 5; i++) { // 第二批线程
                Thread t = new Thread(new SecondBatchWorker(latch));
                t.start();
            }
            // 注意这里也是演示目的的逻辑，并不是推荐的协调方式
            while (latch.getCount() != 1) { // 仅剩一个名额的时候停止等待
                Thread.sleep(1000L);
            }
            System.out.println("第一批全部执行执行完成！");
            latch.countDown(); // 填补完最后一个名额，第二批才会开始执行
        }
    }
    
    class FirstBatchWorker implements Runnable {
    
        private CountDownLatch latch;
    
        public FirstBatchWorker(CountDownLatch latch) {
            this.latch = latch;
        }
    
        @Override
        public void run() {
            System.out.println("第 1 批 executed!");
            latch.countDown(); // latch 次数减一
        }
    }
    
    class SecondBatchWorker implements Runnable {
    
        private CountDownLatch latch;
    
        public SecondBatchWorker(CountDownLatch latch) {
            this.latch = latch;
        }
    
        @Override
        public void run() {
            try {
                latch.await(); // 等待 -> latch 没达到足够的次数
                System.out.println("第 2 批 executed!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

CountDownLatch 的调度方式相对简单，后一批次的线程进行 await，等待前一批 countDown 足够多次。这个例子也从侧面体现出了它的局限性，虽然它也能够支持 10 个人排队的情况，但是因为不能重用，如果要支持更多人排队，就不能依赖一个 CountDownLatch 进行了。其编译运行输出如下：

![1ed45fc8ed444d49a3aa5528dc4538a9.png][]

在实际应用中的条件依赖，往往没有这么别扭，CountDownLatch 用于线程间等待操作结束是非常简单普遍的用法。通过 countDown/await 组合进行通信是很高效的，通常不建议使用例子里那个循环等待方式。

如果用 CyclicBarrier 来表达这个场景呢？我们知道 CyclicBarrier 其实反映的是线程并行运行时的协调，在下面的示例里，从逻辑上，5 个工作线程其实更像是代表了 5 个可以就绪的空车，而不再是 5 个乘客，对比前面 CountDownLatch 的例子更有助于我们区别它们的抽象模型，请看下面的示例代码：// 注意：使用 CyclicBarrier 时，parties 数量应该与线程数量保持一致，一个跑道，一个线程

import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * 可重用的同步阻塞
     */
    public class CyclicBarrierSample {
    
        public static void main(String[] args) throws InterruptedException {
            // 批量等待
            CyclicBarrier barrier = new CyclicBarrier(5, new Runnable() {
                @Override
                public void run() {
                    System.out.println("Action...GO again!");
                }
            });
            for (int i = 0; i < 5; i++) { // 开启 5 个线程
                Thread t = new Thread(new CyclicWorker(barrier));
                t.start();
                Thread.sleep(100);
            }
        }
    
        static class CyclicWorker implements Runnable {
    
            private CyclicBarrier barrier;
    
            public CyclicWorker(CyclicBarrier barrier) {
                this.barrier = barrier;
            }
    
            @Override
            public void run() {
                try {
                    for (int i = 0; i < 3; i++) { // 分 3 个批次
                        System.out.println(Thread.currentThread().getName() + "-" + i + "-" + "准备占用名额");
                        barrier.await(); // 阻塞等待 -> 循环被阻塞
                    }
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

为了让输出更能表达运行时序，我使用了 CyclicBarrier 特有的 barrierAction，当屏障被触发时，Java 会自动调度该动作。因为 CyclicBarrier 会自动进行重置，所以这个逻辑其实可以非常自然的支持更多排队人数。其编译输出如下：

![d31c85dc8168498db0b3dd297f76b52b.png][]

Java 并发类库还提供了[Phaser][]，功能与 CountDownLatch 很接近，但是它允许线程动态地注册到 Phaser 上面，而 CountDownLatch 显然是不能动态设置的。Phaser 的设计初衷是，实现多个线程类似步骤、阶段场景的协调，线程注册等待屏障条件触发，进而协调彼此间行动，具体请参考这个[例子][Link 1]。

当 parties 值不等于线程数时的场景参考文章：

[【线程 锁】栅栏 CyclicBarrier的构造函数入参parties详解\_云川之下的博客-CSDN博客\_cyclicbarrier 构造函数][_ CyclicBarrier_parties_-CSDN_cyclicbarrier]

[ca7fd46a90234806a1047738d627d9e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/325a4dde7348484ba4f026993ec96f20.png
[4d146379267145a885877034cfd6722a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/bbf98cae15d34687beb046d9068bcd61.png
[1ed45fc8ed444d49a3aa5528dc4538a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/dabc22993e2f4b939e45a15ada8315f1.png
[d31c85dc8168498db0b3dd297f76b52b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/607b171a6d33484f9a59e7f6c7e2fea3.png
[Phaser]: https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Phaser.html
[Link 1]: https://www.baeldung.com/java-phaser
[_ CyclicBarrier_parties_-CSDN_cyclicbarrier]: https://blog.csdn.net/m0_45406092/article/details/109598392