【并发编程系列7】CountDownLatch，CyclicBarrier，Semaphore实现原理分析

CountDownLatch

CountDownLatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完毕再执行。

CountDownLatch使用示例

package com.zwx.concurrent.jucUtil;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        new Thread(()->{
            System.out.println("----------");
            countDownLatch.countDown();
        }).start();
        countDownLatch.await();
        System.out.println("==========");
    }
}

CountDownLatch的构造函数接收一个int类型的参数作为计数器，可以视自己的需求设置一个合法的int数值，执行一次countDown()方法，计数器就会减1，await()方法会阻塞线程，直到线程计数减少为0才会继续运行。如果说为了防止某一个线程卡死导致await()一直阻塞的话，也可以调用await(long timeout, TimeUnit unit)方法设定超时时间，到达超时时间之后，即使计数器不到0，也可以继续执行后面的代码。

CountDownLatch源码分析

CountDownLatch(count)

这个方法比较简单，就是维护了一个计数器：


在这里插入图片描述

我们可以看到，调用了Sync类的构造器，而Sync继承了AbstractQueuedSynchronizer类，最终实际上是设置到AbstractQueuedSynchronizer中state属性上，state属性我们前面的AQS文章中提到过，0表示无锁状态，>=1之后表示加锁次数，所以这里的计数器就相当于加锁了N次。


在这里插入图片描述

CountDownLatch#await()

进入await()方法


在这里插入图片描述

然后继续调用了sync的acquireSharedInterruptibly(arg)方法：


如果被中断

1325行的if判断就是判断当前计数器state是不是等于0了，最终调用的是CountDownLatch中的内部类Sync的tryAcquireShared(arg)方法：


在这里插入图片描述

如果state!=0，那就返回-1，返回-1之后需要阻塞线程，所以就会继续执行之后的方法doAcquireSharedInterruptibly(arg)方法。

AQS#doAcquireSharedInterruptibly(arg)

在这里插入图片描述

注意一下999行，这里也是将当前线程封装成一个节点，并构建一个AQS同步队列，前面我们分析重入锁的时候提到了AQS有两种功能，一个是独占，一个就是共享，前面讲过的ReentrantLock中就是以独占模式封装的Node，而这里是以共享模式构建。
共享模式和独占模式在对象中表现出来的区别我们可以进入Node类看一下：


在这里插入图片描述
在这里插入图片描述

所以独占和共享模式构建的节点唯一区别就是共享节点中的nextWaiter不为空（另外还有Condition队列中的nextWaiter也不为空）。

这个方法中前面的一些逻辑AQS中分析过来，这里就不重复分析，这时候我们进来r>=0肯定是不成立的，所以会走到后面的线程挂起，挂起之后线程就阻塞了，那么阻塞了就一定需要被唤醒，所以我们猜测上文示例中的countDown()不但是将计数器减1，肯定还会有判断当减少到0的时候需要唤醒线程。

CountDownLatch#countDown()

调用之后进入CountDownLatch

调用的是sync类中的方法releaseShared(arg)，注意这里固定传的是1，因为调用一次countDown()方法计数减1。

AQS#releaseShared(arg)

在这里插入图片描述

这里做了一个if判断，尝试是否可以释放，如果可以释放之后再执行释放，我们进入tryReleaseShared(arg)方法中一窥究竟。

CountDownLatch#tryReleaseShared(releases)

在这里插入图片描述

注意上面是一个死循环，只有两种情况可以跳出循环，一种就是当前state已经等于0，另一种就是CAS成功，也就是说减1成功。
如果返回false，就说明还需要阻塞等待其他线程；如果返回的是true，就会直接后面的doReleaseShared()方法。

AQS#doReleaseShared()

这个方法主要就是通过一个循环将head节点唤醒，因为中途可能会被其他线程唤醒了或者也可能加入了新节点，所以需要通过一个死循环来确保释放成功


在这里插入图片描述

回到AQS#doAcquireSharedInterruptibly(arg)

在这里插入图片描述

上面await()方法的线程阻塞在1014这个if条件这里，唤醒之后如果没有被中断过，那么会继续执行for循环，这时候r>=0肯定成立了，所以会进入setHeadAndPropagate(Node,int)方法，去依次传播所有需要唤醒的节点

AQS#setHeadAndPropagate(Node,int)

在这里插入图片描述

这里注意到参数中的Node是head节点的下一个节点，所以这里要做的是把第二个节点替换成Node节点，然后执行同一个方法doReleaseShared()方法去唤醒头节点，唤醒之后会回到上面的for循环，继续唤醒后一个节点，直到全部线程均被唤醒。

CyclicBarrier

CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一 组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会 开门，所有被屏障拦截的线程才会继续运行。CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景

CyclicBarrier使用示例1

package com.zwx.concurrent.jucUtil;

import java.util.concurrent.CyclicBarrier;

public class CyclicbarrierDemo {
    static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
    public static void main(String[] args) {
        new Thread(()-> {
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("我是线程t1");
        },"t1").start();
        new Thread(()-> {
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("我是线程t2");
        },"t2").start();

        System.out.println("主线程==end");
    }
}

输出结果:


在这里插入图片描述

这个t1和t2的输出结果是随机的。

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier（int parties，Runnable barrier- Action），用于在线程到达屏障时，优先执行barrierAction。

CyclicBarrier使用示例2

package com.zwx.concurrent.jucUtil;

import java.util.concurrent.CyclicBarrier;

public class CyclicbarrierDemo2 {
    static CyclicBarrier cyclicBarrier = new CyclicBarrier(2,new MyThread());
    public static void main(String[] args) {
        new Thread(()-> {
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("我是线程t1");
        },"t1").start();
        new Thread(()-> {
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("我是线程t2");
        },"t2").start();

        System.out.println("主线程==end");
    }

    static class MyThread extends Thread {
        @Override
        public void run() {
            System.out.println("do something");
        }
    }
}

输出结果：


在这里插入图片描述

我们可以看到，在线程t1和t2输出前会先输出自定义线程的信息。

CyclicBarrier实现原理

CyclicBarrier 相比 CountDownLatch 来说，要简单很多，源码实现是基于 ReentrantLock 和 Condition 的组合使用

CyclicBarrier源码分析

CyclicBarrier(parties)

进入CyclicBarrier默认构造器：


在这里插入图片描述

可以发现，最终其实还是调用的CyclicBarrier（int parties，Runnable barrier- Action）构造器:


在这里插入图片描述

注意了，构造CyclicBarrier对象时，初始化了多少个parties，则必须对应有parties个线程调用await()方法，否则线程不会往后执行。

CyclicBarrier#await()

在这里插入图片描述

调用了dowait(timed,nanos)方法，第一个参数false表示未设置超时时间，后面表示纳秒数，因为await还有另一个对应的方法带上超时时间：await(long,timeunit)，这个方法中调用dowait(timed,nanos)方法时第一个参数就会是true，然后带上超时时间，表示到了设定时间之后线程就不会被阻塞，会继续往后执行。

CyclicBarrier#dowait()

/**
     * Main barrier code, covering the various policies.
     * 主要屏障代码，覆盖了各种策略
     */
    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;//定义一个重入锁:private final ReentrantLock lock = new ReentrantLock();
        lock.lock();
        try {
            //同一个屏障初始进来时属于同一代或者说一个周期，构建一个"代"(Generation)对象，同一个Generation表示同一代
            final Generation g = generation;//Generation中设置了broke=false，表示屏障没有损坏

            if (g.broken)//如果broken=true表示当前屏障被损坏了，抛出异常
                throw new BrokenBarrierException();

            if (Thread.interrupted()) {//如果线程被中断过
                breakBarrier();//设置屏障为损坏状态并唤醒所有持有锁的线程
                throw new InterruptedException();//抛出中断异常
            }

            int index = --count;//未调用await()方法的线程计数-1
            if (index == 0) {//如果屏障数为0,(表示所有线程都到达await()方法)
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();//表示到达屏障之后，如果我们有设置barrierCommand，则优先执行
                    ranAction = true;
                    //执行到这里的时候，说明所有线程都到了await()方法，且设置的barrierCommand也已经执行完了
                    //接下来要做的事情就是换代(所以CyclicBarrier是通过换代的方式实现重新计数的)
                    //换代之后相当于进入一个新的周期，所有线程在后续中又可以通过await()阻塞一次
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)//如果ranAction = false说明当前屏障还有流程没执行完，所以需要屏障设置会损坏状态
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            //死循环等到count=0，调用breakBarrier方法(表示屏障有问题的场景),中断或者超时
            for (;;) {
                try {
                    if (!timed)
                        //private final Condition trip = lock.newCondition();
                        trip.await();//即Condition队列的await()阻塞，相当于把线程加入到Condition队列中阻塞
                    else if (nanos > 0L)//超时时间大于0
                        nanos = trip.awaitNanos(nanos);//阻塞指定时间
                } catch (InterruptedException ie) {
                    //如果当前屏障没有换代，也没有损坏，那么就设置为损坏状态之后再抛出中断异常
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();
                    }
                }

                if (g.broken)//如果屏障已经被损坏了
                    throw new BrokenBarrierException();

                if (g != generation)//如果发现已经换代了，就不继续循环了，直接返回就好了
                    return index;//返回当前还有多少个线程没有执行await()方法

                if (timed && nanos <= 0L) {//表示超时时间到了
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

这里的方法看起来很长，但其实除了一系列的判断，并没有多代码，结合注释，如果前面理解了ReentrantLock和Condition队列的话，应该非常好看懂，里面调用的其他一些子方法这里也不做单独介绍。

使用CyclicBarrier注意事项

1、对于指定计数值 parties，若由于某种原因，没有足够的线程调用 CyclicBarrier 的await()方法，则所有调用 await 的线程都会被阻塞；
2、若有多余线程执行了await()方法，那么最后一个到达屏障的线程会被阻塞
3、通过 reset 重置计数，会使得进入 await 的线程出现BrokenBarrierException；我们可以通过捕获异常重新处理业务逻辑
4、如果采用是 CyclicBarrier(int parties, Runnable barrierAction) 构造方法，执行 barrierAction 操作的是最后一个到达的线程。

CountDownLatch和CyclicBarrier区别

  • CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重置。所以CyclicBarrier能处理更为复杂的业务场景。例如，如果计算发生错误，可以重置计数 器，并让线程重新执行一次。

Semaphore

Semaphore也就是我们常说的信号灯，Semaphore可以控制同时访问的线程个数，通过 acquire 获取一个许可，如果没有就等待，通过 release 释放一个许可。有点类似限流
的作用。叫信号灯的原因也和他的用处有关，比如某商场就 5 个停车位，每个停车位只能停一辆车，如果这个时候来了 10 辆车，必须要等前面有空的车位才能进入。

Semaphore使用示例

package com.zwx.concurrent.jucUtil;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore=new Semaphore(2);
        for(int i=1;i<=5;i++){
            new Car(i,semaphore).start();
        }
    }
}

class Car extends Thread{
    private int num;
    private Semaphore semaphore;

    public Car(int num, Semaphore semaphore) {
        this.num = num;
        this.semaphore = semaphore;
    }
    @Override
    public void run() {
        try {
            semaphore.acquire();//获取一个许可
            System.out.println("第"+num+"辆车进来了");

            TimeUnit.SECONDS.sleep(2);
            System.out.println("第"+num+"辆车出去了");
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果：


在这里插入图片描述

我们可以看到，一开始进来2辆车之后就阻塞了，后面必须出去一辆车才能进来一辆车。

Semaphore源码分析

Semaphore(permits)

Semaphore实际上和CountDownLatch实现非常相似，构造器最终的结果也是调用Sync类中将AbstractQueuedSynchronizer类中的state属性设置为permits:


在这里插入图片描述

上图说明默认构造的是非公平锁，但是还提供了另一个构造器由我们自己决定使用非公平锁还是公平锁，构造器最终是调用的下面这个方法设置state属性：


在这里插入图片描述

Semaphore#acquire()

这个方法和CountDownLatch中的await()调用的是同一个方法，这里就不再做分析

Semaphore#release()

这个方法调用的和CountDownLatch中的countDown()调用的也是同一个方法：


在这里插入图片描述

最终调用的tryReleaseShared(arg)会和CountDownLatch会有一点点差异：

Semaphore#tryReleaseShared()

在这里插入图片描述

当释放了一个令牌之后，通过将允许的令牌总数+1实现多进来一个线程。

AQS#doReleaseShared()

上面tryReleaseShared()返回true之后，就会去唤醒下一个线程，这个和上面CountDownLatch中的countDown()方法调用的也是同一个方法去唤醒下一个线程。

总结

本篇文章主要介绍了三个常用的工具CountDownLatch，CyclicBarrier，Semaphore，其中，CountDownLatch和CyclicBarrier在一定场景下是可以替换使用的，而Semaphore一般用于限流。

后面并发编程系列将继续介绍JUC包下面队列如ConcurrentLinkedQueue和阻塞队列等相关知识。感兴趣的 请关注我，一起学习进步。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,189评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,577评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,857评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,703评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,705评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,620评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,995评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,656评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,898评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,639评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,720评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,395评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,982评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,953评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,195评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,907评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,472评论 2赞 342