CountDownLatch

主要的功能就是通过await()方法来阻塞线程,然后等待计数器减少到0了,再唤起那些等待的线程继续;即你想要某些线程等待另一些线程执行完再执行,就可以使用CountDownLatch。
等待线程执行await,等待直到同步状态state被被等待线程减为0,唤醒等待线程。
先来看看用法:

//三个线程阻塞直到主线程执行完
    public void test1() {
        final CountDownLatch latch = new CountDownLatch(1);

        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + " wait;");
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + " start");
            }).start();
        }

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        latch.countDown();
    }

//主线程阻塞等待直到三个线程都执行完
    public void test2() {
        final CountDownLatch latch = new CountDownLatch(3);

        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + " start");
                latch.countDown();
            }).start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("All end");
    }

源码

构造函数:同步状态state的值在CountDownLatch构造函数中赋值(AQS为同步状态提供了getState,setState方法,它们即没有锁保护,也没用使用CAS,因为某些情况下只需要可见性得到保证即可,所以用volatile修饰)。

    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

之前文章里说过,AQS是JUC之基,各同步器都采用实现一个内部类继承AQS的方式,之前分析的ReentrentLock也是

    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

AQS有独占与共享两种锁,再创建Node是会指定其中一种模式,之前分析过,两种锁最大不同在tryAcquire操作的实现上,独占锁子类实现的tryAcquire返回boolean,代表是否有更改同步状态的资格;共享锁子类实现tryAcquireShared返回int,小于0则放入等待队列中。
CountDownLatch的tryAcquireShared逻辑是同步状态为0时才会返回正数,大于零返回负数;逻辑是等被等待线程执行完才允许你执行。
tryReleaseShared尝试将同步状态减一。

await

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

//定位到AQS里
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

await的逻辑就是如果同步状态大于0,线程就会进入AQS的等待队列中挂起;同步状态等于0则线程不会被阻塞,直接执行;

  1. 注意addWaiter方法,第一个进队列的线程会排在第二个位置,它之前会有个空Node作为头节点,实现在enq()方法中。这里就很好理解为什么解锁时一上来调用的是unparkSuccessor方法(唤醒head.next节点),因为头是个空节点;之后进队列的节点就依此连接并改变前一节点的状态值为SIGNAL。
  2. 还有需要注意的是:CountDownLatch的设计是当同步状态变为零则等待队列中所有节点都将被唤醒,而下面的唤醒操作doReleaseShared()一次只唤醒一个,那之后等待的线程由谁来唤醒?由被唤醒的线程。当一个线程被唤醒后仍在doAcquireSharedInterruptibly的for循环里,它会有一次循环而tryAcquireShared返回值肯定为1,那么setHeadAndPropagate会被调用
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node); //设置新头节点

        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

参数propagate是tryAcquireShared的返回值,这里为1;设置当前节点为新的头节,再次执行doReleaseShared唤醒后继节点。依此往后直到全部唤醒。

举例说明:就以上面的test1()为例
该例子设计是三个线程A,B,C需要等待直到主线程D执行完了再继续执行。假设A,B,C按顺序在队列中等待,此时队列中会有四个节点,头节点是一个空节点。D执行了countDown将同步状态改为0,之后按下面源码的顺序,A会被唤醒在doAcquireSharedInterruptibly中调用setHeadAndPropagate,在之后B被唤醒,再之后C。

countDown

    public void countDown() {
        sync.releaseShared(1);
    }

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
//SIGNAL节点状态代表当前节点有需要唤醒的后继节点
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))  //归零
                        continue;            // loop to recheck cases
                    unparkSuccessor(h); //唤醒head.next节点
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }
    private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
//节点状态重置为0;节点的状态代表不同的行为,0为初始也代表要删除的状态。
            compareAndSetWaitStatus(node, ws, 0); 

        Node s = node.next;
//跳过状态为CANCEL的节点
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread); //唤醒
    }

tryReleaseShared利用循环CAS将同步状态state减1。doReleaseShared唤醒head.next节点。
注意:上面doReleaseShared的for循环里有个ws == 0的判断,会将节点的状态变为PROPAGATE,以我们上面的举例来说,ws似乎不可能为0,那么什么情况下ws==0?节点的状态初始为0,会被后继节点设为SIGNAL(doAcquireSharedInterruptibly的shouldParkAfterFailedAcquire里实现)那么可能的情况就是,线程B调用tryAcquireShared返回-1,即同步状态此时还未为0,之后调用doAcquireSharedInterruptibly——>addWaiter此时节点以入队列,但还未改变它的前一个节点A的状态为SIGNAL,也未挂起自己,这时其他线程抢进来将同步状态减为0,接着执行doReleaseShared,这是便会出现ws == 0的情况,这里会将A的状态改为PROPAGATE,再回到B,B调用shouldParkAfterFailedAcquire将A状态改为SIGNAL并返回false,那么这样A就不会执行parkAndCheckInterrupt将自己挂起,再次循环检查是否轮到自己,否则他会将自己挂起,在队列中等待直到轮到自己。

超时

    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

//方法再AQS里

    public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquireShared(arg) >= 0 ||
            doAcquireSharedNanos(arg, nanosTimeout);
    }
    private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
//最终时间
        final long deadline = System.nanoTime() + nanosTimeout;
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return true;
                    }
                }
                nanosTimeout = deadline - System.nanoTime();
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

返回true,代表在时间范围内同步状态归零;返回false代表超时。
本质上利用LockSupport.parkNanos(this, nanosTimeout);
实现与doReleaseShared相似,由于有超时操作则应该考虑到当时间超了,会对上面说的执行流程有什么影响:超时后线程恢复,线程状态变为RUNNABLE就绪状态,抢到执行权后,先在for循环中判断是否轮到自己执行,同步状态是否变为零;否则直接返回false,则该线程不会再等待,会执行下去。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 213,928评论 6 493
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 91,192评论 3 387
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 159,468评论 0 349
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 57,186评论 1 286
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 66,295评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,374评论 1 292
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,403评论 3 412
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,186评论 0 269
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,610评论 1 306
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,906评论 2 328
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,075评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,755评论 4 337
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,393评论 3 320
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,079评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,313评论 1 267
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,934评论 2 365
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,963评论 2 351

推荐阅读更多精彩内容

  • CountDownLatch 介绍 CountDownLatch是一个同步协助类,允许一个或多个线程等待,直到其他...
    tomas家的小拨浪鼓阅读 3,191评论 0 9
  • ReentrantLock 介绍 一个可重入的互斥锁,它具有与使用{synchronized}方法和语句访问的隐式...
    tomas家的小拨浪鼓阅读 4,047评论 1 4
  • 前言 上一篇文章《基于CAS操作的Java非阻塞同步机制》 分析了非同步阻塞机制的实现原理,本篇将分析一种以非同步...
    Mars_M阅读 4,798评论 5 9
  • 很多人都喜欢说想当年怎么怎么样!无数次的说自己曾经也是拼过的人,无数次的说自己曾经辉煌过。但其实这只是曾经,现在呢...
    乐梅拾坚阅读 318评论 0 0
  • 一不小心,已经下午两点。 昨天一点到顺德了能,已经停止派号;转而顺德私房菜,一半的菜品已卖光;三点到金榜牛奶,最后...
    山楂饴阅读 318评论 0 0