并发相关工具

Myth丶恋晨 2022-12-28 15:27 196阅读 0赞

点击上方“晏霖”，选择“置顶或者星标”

曾经有人关注了我

后来他有了女朋友

我们在并发编程中，经常会使用到一些工具来帮助我们控制线程。本章节就会对CountDownLatch、CyclicBarrier、Semaphore工具的应用进行简单的介绍。

2.11.1 CountDownLatch

CountDownLatch，使用AQS状态表示计数，可以把它看成是一个计数器，源码注释第一句话：A synchronization aid that allows one or more threads to wait until，翻译过来就是：允许一个或多个线程等待。好，让我们揭开CountDownLatch的面纱。当我们打开CountDownLatch源码时，可以看到300多行代码，80%是注释。CountDownLatch类中官方的解释是：允许一个或多个线程等待，在其他线程中执行的一组操作完成。用给定的count初始化。方法块\{@link\#await await\}直到当前计数达到由于调用\{@link\#countDown\}方法为零，之后释放所有等待的线程以及随后的任何调用。

使用CountDownLatch最重要2个方法，一个是countDown();函数，另一个是await();函数。countDown是递减锁存器的计数，如果计数到达零，则释放所有等待的线程。如果当前计数大于零，则将计数减少1。await方法可以使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断 ， 如果当前的计数为零，则此方法立即返回。让我们利用一个demo介绍一下CountDownLatch如何的应用，请看示例代码2-48所示。

代码清单2-48 CountDownLatchDemo.java

public class CountDownLatchDemo {
        private static CountDownLatch countDownLatch = new CountDownLatch(5);
        public static void main(String[] args) throws InterruptedException {
            // 创建定长线程池
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            for (int i = 1; i <= 5; i++) {
                executorService.execute(() -> {
                    int consumeTime = new Random().nextInt(10) + 1;
                    String name = Thread.currentThread().getName();
                    System.out.println(name + ":我要" + consumeTime + "秒才能完成任务!");
                    try {
                        Thread.sleep(consumeTime * 1000);
                        System.out.println(name + "完成任务啦!");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        // 倒计时减一
                        // 注.countDown()方法是线程安全的
                        countDownLatch.countDown();
                    }
                });
            }
            // 当countDownLatch的值不为0时,此线程阻塞，继续等待;直到countDownLatch的值=0时,此线程才往下执行
            // 这里是.await()方法,而不是.wait()方法
            countDownLatch.await();
            // 关闭线程池
            executorService.shutdown();
            System.out.println(Thread.currentThread().getName() + ":终于该本线程出手了!");
        }
    }

打印输出结果。

pool-1-thread-3:我要8秒才能完成任务!

pool-1-thread-2:我要4秒才能完成任务!

pool-1-thread-1:我要6秒才能完成任务!

pool-1-thread-4:我要5秒才能完成任务!

pool-1-thread-5:我要9秒才能完成任务!

pool-1-thread-2完成任务啦!

pool-1-thread-4完成任务啦!

pool-1-thread-1完成任务啦!

pool-1-thread-3完成任务啦!

pool-1-thread-5完成任务啦!

main:终于该本线程出手了!

2.11.2 CyclicBarrier

CyclicBarrier俗称栅栏，我们也可以按翻译过来的字面含义，分为可循环的(cyclic)的屏障(barrier)。他它允许一组线程互相等待，直到到达某个公共屏障点，然后释放这些线程，重置屏障点继续等待，知道所有要执行的线程都执行完毕。为了更容易理解这个栅栏的含义，我做一个比喻，目前有100个人要坐车去另一个地方，每个车可以装10个人，那么这个屏障点就是车里坐满了10个人，然后发车，接着马上重置，然后让后面10个人在坐上车，以此类推，直到这100个人都执行完毕。

CyclicBarrier可以用两个构造方法调用，一个是默认都构造方法（单个参数，只设置屏障点），一个是用于线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景，但一般选用单个参数的。

CyclicBarrier最重要的方法就是await，调用await方法的线程告诉CyclicBarrier有一个线程已经到达同步点，然后当前线程被阻塞。直到parties（设置的屏障数量）个参与线程调用了await方法。CyclicBarrier同样提供带超时时间的await方法。

下面用一个简单的代码案例来说明如何使用CyclicBarrier，请看示例代码2-49所示。

代码清单2-49 CountDownLatchDemo.java

public class CyclicBarrierDemo {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newCachedThreadPool();
            CyclicBarrier cyclicBarrier = new CyclicBarrier(9);
            for (int i = 1; i <= 18; i++) {
                executorService.execute(new Runnable() {
                    @Override
                    public void run() {
                        System.out.println(Thread.currentThread().getName() + "开始等待其他线程");
                        try {
                            cyclicBarrier.await();
                            System.out.println(Thread.currentThread().getName() + "业务逻辑执行完毕");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (BrokenBarrierException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
            executorService.shutdown();
        }
    }

使用了一个缓存线程执行18个任务，任务要求每次处理的线程达到9个时候就释放一次，不足9个等待。

控制台输出结果如下。

pool-1-thread-1开始等待其他线程

pool-1-thread-2开始等待其他线程

pool-1-thread-3开始等待其他线程

pool-1-thread-4开始等待其他线程

pool-1-thread-5开始等待其他线程

pool-1-thread-6开始等待其他线程

pool-1-thread-7开始等待其他线程

pool-1-thread-8开始等待其他线程

pool-1-thread-9开始等待其他线程

pool-1-thread-10开始等待其他线程

pool-1-thread-9业务逻辑执行完毕

pool-1-thread-1业务逻辑执行完毕

pool-1-thread-2业务逻辑执行完毕

pool-1-thread-11开始等待其他线程

pool-1-thread-3业务逻辑执行完毕

pool-1-thread-1开始等待其他线程

pool-1-thread-9开始等待其他线程

pool-1-thread-4业务逻辑执行完毕

pool-1-thread-2开始等待其他线程

pool-1-thread-5业务逻辑执行完毕

pool-1-thread-6业务逻辑执行完毕

pool-1-thread-7业务逻辑执行完毕

pool-1-thread-8业务逻辑执行完毕

pool-1-thread-12开始等待其他线程

pool-1-thread-13开始等待其他线程

pool-1-thread-3开始等待其他线程

pool-1-thread-4开始等待其他线程

pool-1-thread-4业务逻辑执行完毕

pool-1-thread-10业务逻辑执行完毕

pool-1-thread-11业务逻辑执行完毕

pool-1-thread-1业务逻辑执行完毕

pool-1-thread-9业务逻辑执行完毕

pool-1-thread-2业务逻辑执行完毕

pool-1-thread-12业务逻辑执行完毕

pool-1-thread-13业务逻辑执行完毕

pool-1-thread-3业务逻辑执行完毕

2.11.3 Semaphore

Semaphore 翻译成字面意思为 信号量，Semaphore 可以控制同时访问的线程个数，通过

acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。

Semaphore 类中比较重要的几个方法：

1．public void acquire(): 用来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可。

2．public void acquire(int permits):获取 permits 个许可。

3．public void release() \{ \} :释放许可。注意，在释放许可之前，必须先获获得许可。

4．public void release(int permits) \{ \}:释放 permits 个许可。

这4 个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法。

1．public boolean tryAcquire():尝试获取一个许可，若获取成功，则立即返回 true，若获取失败，则立即返回 false。

2．public boolean tryAcquire(long timeout, TimeUnit unit):尝试获取一个许可，若在指定的时间内获取成功，则立即返回 true，否则则立即返回 false。

3．public boolean tryAcquire(int permits):尝试获取 permits 个许可，若获取成功，则立即返回 true，若获取失败，则立即返回 false。

4．public boolean tryAcquire(int permits, long timeout, TimeUnit unit): 尝试获取 permits个许可，若在指定的时间内获取成功，则立即返回 true，否则则立即返回 false。

还可以通过 availablePermits()方法得到可用的许可数目。

下面我们用代码模拟以下案例。

游乐园有100个人买了攀爬的票,但是攀爬的绳子只准备了20条(这就是信号量),那么最多同时只能有20个人享受攀爬,其余的人只有在下面等待,只有当有人攀爬结束后,才会有空的位置让出来给哪些等待的人。

请看示例代码2-50所示。

代码清单2-50 SemaphoreDemo.java

public class SemaphoreDemo {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newCachedThreadPool();
            // 获取型号量实例(设置允许最大线程数为20个)
            Semaphore semaphore = new Semaphore(20);
            for (int i = 1; i <= 100; i++) {
                // 使用lambel表达式简单实现Runnable接口的run方法
                executorService.execute(() -> {
                    try {
                        // 申请攀岩绳索(即:获取资格)  -> 如果没有获取到资格那么就一直等下去,直到获取到了资格为止
                        semaphore.acquire();
                        int consumeTime = new Random().nextInt(10) + 1;
                        String name = Thread.currentThread().getName();
                        System.out.println(name + "获取了攀岩资格，开始攀岩!");
                        Thread.sleep(consumeTime * 1000);
                        System.out.println(name + "攀岩结束！");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        // 让出绳索(即:释放资格)
                        semaphore.release();
                    }
                });
            }
            executorService.shutdown();
        }
    }

篇幅原因，省略输出结果，读者可自行去码云克隆本书的代码执行。

2.11.4 CountDownLatch、Semaphore、 CyclicBarrier的区别

l CountDownLatch 和 CyclicBarrier 都能够实现线程之间的等待，只不过它们侧重点不

l 同；CountDownLatch 一般用于某个线程 A 等待若干个其他线程执行完任务之后，它才执行；而 CyclicBarrier 一般用于一组线程互相等待至某个状态，然后这一组线程再同时

l 执行；另外，CountDownLatch 是不能够重用的，而 CyclicBarrier 是可以重用的。

Semaphore 其实和锁有点类似，它一般用于控制对某组资源的访问权限。

![de154207d4a3ca9495c955110e92a490.png][]

胖虎

热 爱 生 活 的 人

终 将 被 生 活 热 爱

我在这里等你哟！

![4f73367d805e523e1aaf879ec816572d.png][]

![674a23941c0fceeb95de395092da6d16.gif][]

[de154207d4a3ca9495c955110e92a490.png]: /images/20221120/257f44f0aee042b486f60354efd1d845.png
[4f73367d805e523e1aaf879ec816572d.png]: https://img-blog.csdnimg.cn/img_convert/4f73367d805e523e1aaf879ec816572d.png
[674a23941c0fceeb95de395092da6d16.gif]: https://img-blog.csdnimg.cn/img_convert/674a23941c0fceeb95de395092da6d16.gif