zoukankan      html  css  js  c++  java
  • CountDownLatch源码探究 (JDK 1.8)

    CountDownLatch能够实现让线程等待某个计数器倒数到零的功能,之前对它的了解也仅仅是简单的使用,对于其内部如何实现线程等待却不是很了解,最好的办法就是通过看源码来了解底层的实现细节。CountDownLatch的源码并不是很复杂,因为其核心的功能是依赖AbstractQueuedSynchronizer(下文简称AQS)来实现的。CountDownLatch常用的方法很少,但是因为涉及到AQS,逻辑有些绕,要理清中间的逻辑稍微要费一些时间。

    1.内部类Sync

    CountDownLatch的核心功能是通过内部类Sync实现的,这个类继承了AQS

    	private static final class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 4982264981922014374L;
    
            //构造器,根据传入的整数初始化状态字段state
            Sync(int count) {
                setState(count);
            }
    
            int getCount() {
                return getState();
            }
    
            //tryAcquireShared唯一的作用是查看状态字段是不是等于0
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
    
            protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
                //自旋,在两种条件下会退出自旋:a)state字段已经为0;b)线程成功地将state字段减1
                for (;;) {
                    int c = getState();
                    //如果state已经为0,就返回false
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    //从下面的语句可以看到,只有当state=0才会返回true
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
        }
    

    2.构造器

    CountDownLatch只有一个构造器,在构造器中会初始化sync字段,结合Sync类的定义可知,构造器的唯一工作是将state字段初始化为传入的参数:

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

    3.节点状态waitStatus

    等待的线程会构造成节点放在等待队列中,节点的状态waitStatus有如下几种:

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;
    

    注意,在CountDownLatch中并没有用到CONDITION状态,因此后文将会直接忽略该状态,当waitStatus > 0时,指的就是CANCELLED状态。

    4.核心方法

    • await()
      当计数器不等于0时,await()方法会让当前线程挂起,该方法调用了AQS类的acquireSharedInterruptibly方法,如下:
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
        public final void acquireSharedInterruptibly(int arg)  throws InterruptedException {
    	    if (Thread.interrupted())
    	        throw new InterruptedException();
    	    //显然,tryAcquireShared方法只有在state=0时才返回1,表示计数器已归零,此时方法直接返回,被阻塞的线程就可以继续执行
    	    if (tryAcquireShared(arg) < 0)
    	        doAcquireSharedInterruptibly(arg);
        }
    

    通常,调用await()的线程在执行到acquireSharedInterruptibly方法时,计数器并不为0,那么当前线程就需要执行doAcquireSharedInterruptibly方法中的阻塞逻辑了。由于该方法内部调用了三个主要方法:addWaitershouldParkAfterFailedAcquireparkAndCheckInterrupt,在解析的过程中难免会穿插对这些方法的介绍,从而引入跳跃性。为了避免跳跃性引发的阅读和理解上的困难,这里准备先介绍addWaiter方法。

    • addWaiter
        private Node addWaiter(Node mode) {
        	//将当前线程构造成一个Node节点
            Node node = new Node(Thread.currentThread(), mode);
            //获取尾节点
            Node pred = tail;
            //尾节点不为空,说明队列已完成初始化
            if (pred != null) {
            	//将node节点放到对尾,这里的做法是先将node的prev指针指向尾节点,然后通过原子操作将新添加的node更新成尾节点,成功的话addWaiter方法结束
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                	//原子操作成功的话,更新原尾节点的next指针
                    pred.next = node;
                    return node;
                }
            }
            //执行到这里有两种情况:1)尾节点为空,即队列还没初始化;2)队列已初始化,但是上文将node节点设置成尾节点失败,此时node节点还没有真正添加进队列
            enq(node);
            return node;
        }
    
        private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                //如果队列还没初始化,则先初始化,做法是将一个空节点作为头结点,然后让尾节点也指向这个空节点
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    //这里会一直自旋,直到成功地将node节点更新成尾节点
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    
    

    addWaiter方法的主要作用就是将当前线程添加到等待队列的队尾,如果队列还没初始化,则先初始化,enq方法使用自旋避免入队失败。

    • doAcquireSharedInterruptibly
      接下来正式开始介绍doAcquireSharedInterruptibly方法,源码如下:
        private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            //将当前线程添加到等待队列,注意参数是Node.SHARED,下文会用到
            final Node node = addWaiter(Node.SHARED);
            //该字段在state=0时才会被设置为false
            boolean failed = true;
            try {
            	//又是自旋,该自旋的终止条件有两种:1)state=0,计数器正常结束,执行return语句返回;2)线程响应中断异常,跳出自旋
                for (;;) {
                	//获取node的前驱节点
                    final Node p = node.predecessor();
                    //如果前驱节点是头结点,则执行if代码块的逻辑
                    //这段逻辑说明,每次都是头结点之后的有效节点才可以执行,其他节点即使有机会醒来也不会执行下面的逻辑,而是会在后面继续挂起
                    if (p == head) {
                    	//获取state字段的状态,如果state=0则返回1,否则返回-1
                        int r = tryAcquireShared(arg);
                        //r>=0,说明计数器结束了,需要唤醒阻塞的线程
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            //计数器正常结束时,会将failed设置为false,避免执行finally中的语句
                            failed = false;
                            return;
                        }
                    }
                    //执行到这里有两种情况:1)前置节点不是头结点;2)前置节点是头结点,但state!=0,计数器还没结束。
                    //真正的阻塞逻辑在parkAndCheckInterrupt方法里
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
            	//如果线程被中断,那么failed=true,执行cancelAcquire方法
                if (failed)
                    cancelAcquire(node);
            }
        }
    

    doAcquireSharedInterruptibly先通过addWaiter方法将当前线程添加到等待队列尾部,然后开始自旋。如果state字段不为0,那么会执行到末尾的条件语句:

    	if (shouldParkAfterFailedAcquire(p, node) &&
            parkAndCheckInterrupt())
            throw new InterruptedException();
    

    先来看看shouldParkAfterFailedAcquire干了些什么:

        //注意pred是node的前驱节点
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            //如果已经是SIGNAL状态,则直接返回true
            if (ws == Node.SIGNAL)
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
                return true;
            //ws>0只能是cancelled状态,此时通过修改指针将这些cancelled的节点从队列删除
            if (ws > 0) {
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                /*
                 * waitStatus must be 0 or PROPAGATE.  Indicate that we
                 * need a signal, but don't park yet.  Caller will need to
                 * retry to make sure it cannot acquire before parking.
                 */
                //如果前驱节点的状态既不是SIGNAL,也不是CANCELLED,那么只可能是0或者PROPAGATE,就把前驱节点的状态更新为 Node.SIGNAL。注意:1)CONDITION状态在CountDownLatch中并没有用到;2)节点新建的时候状态都是0,是在这里才被修改成了SIGNAL
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
    

    shouldParkAfterFailedAcquire的主要作用是根据前驱节点的状态进行操作:

    • 如果前驱节点是SIGNAL,直接返回true,其他情况返回false;
    • 如果前驱节点是CANCELLED,则从该节点开始向前查找连续的CANCELLED节点,将这些节点删除;
    • 除上面的两种情况外,将节点状态改为SIGNAL。
      之前对节点的SIGNAL状态是怎么来的一直有点迷糊,看了上面的代码才发现是在最后一个else分支中设置的。从shouldParkAfterFailedAcquire源码了解到,该方法只有在前驱节点状态是SIGNAL时才返回true,此时才有机会执行parkAndCheckInterrupt方法。parkAndCheckInterrupt是真正让线程挂起的地方,来看看其源码:
        private final boolean parkAndCheckInterrupt() {
        	//线程最终会阻塞在这里,线程恢复之后也将从这里继续执行
            LockSupport.park(this);
            return Thread.interrupted();
        }
    

    parkAndCheckInterrupt方法借助LockSupport实现线程阻塞,被阻塞的线程在被唤醒后会返回当前线程的中断状态(注意Thread.interrupted()会清除线程的中断状态)。好了,到这里整个逻辑就比较清楚了,如果线程是正常被唤醒(即state=0),那么parkAndCheckInterrupt返回falsedoAcquireSharedInterruptibly方法会接着自旋一次,这里再次将自旋代码贴出:

        for (;;) {
        	//获取node的前驱节点
            final Node p = node.predecessor();
            //如果前驱节点是头结点,则执行if代码块的逻辑
            if (p == head) {
            	//获取state字段的状态,如果state=0则返回1,否则返回-1
                int r = tryAcquireShared(arg);
                //r>=0,说明计数器结束了,需要唤醒阻塞的线程
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            //执行到这里说明state!=0,真正的阻塞逻辑在parkAndCheckInterrupt方法里
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    

    那么setHeadAndPropagate方法做了些什么事呢,看看它的源码(删掉了源码中的注释):

        //回忆一下,显然propagate=1,node是当前醒来的线程对应的节点
        private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; // Record old head for check below
            //把node设置为头结点
            setHead(node);
            //此时propagate > 0的条件已经满足,直接执行if代码块的逻辑,注意if条件第一行的h是原来的头结点,第二行是新的头结点
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                //如果没有下一个节点,或者下一个节点的isShared返回true,就释放。还记得吗,在构造新节点的时候addWaiter的参数是Node.SHARED,这里就是判断这个字段
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }
    
        private void setHead(Node node) {
            head = node;
            node.thread = null;
            node.prev = null;
        }
    

    接下来看一下doReleaseShared是如何实现的:

        private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            for (;;) {
            	//获取头结点
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    //如果头结点的状态是SIGNAL,那么会将其状态修改为0,该步骤一直自旋直到成功为止
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        //成功修改头结点的状态后,会执行下面这个方法
                        unparkSuccessor(h);
                    }
                    //如果头结点状态已经改成0了,就再次将其状态更新为Node.PROPAGATE,目的是???
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }
    

    可以看到,doReleaseShared每次只会唤醒头结点之后的有效节点,头结点的状态成功更新为0后,会执行unparkSuccessor方法的逻辑,该方法源码如下:

        private void unparkSuccessor(Node node) {
            /*
             * If status is negative (i.e., possibly needing signal) try
             * to clear in anticipation of signalling.  It is OK if this
             * fails or if status is changed by waiting thread.
             */
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            /*
             * Thread to unpark is held in successor, which is normally
             * just the next node.  But if cancelled or apparently null,
             * traverse backwards from tail to find the actual
             * non-cancelled successor.
             */
            //获取后继节点
            Node s = node.next;
            //如果没有后继节点,或者后继节点是CANCELLED状态,则执行下面的代码块
            if (s == null || s.waitStatus > 0) {
                s = null;
                //从队列末尾向开头遍历,找到靠近头结点的第一个不为CANCELLED状态的节点
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            //找到这样的非CANCELLED节点,就将其唤醒
            if (s != null)
                LockSupport.unpark(s.thread);
        }
    

    unparkSuccessor的主要工作是将头结点后面第一个非CANCELLED状态的节点所对应的线程唤醒。

    • cancelAcquire
      到目前为止,并没有发现CANCELLED状态是在哪里设置,因为还有一个方法没有分析。doAcquireSharedInterruptibly中的finally语句块会处理线程被中断的情况,执行的是cancelAcquire方法的逻辑,其源码如下:
        private void cancelAcquire(Node node) {
            // Ignore if node doesn't exist
            if (node == null)
                return;
            //线程中断后,将其对应的节点中保存的线程清空
            node.thread = null;
    
            // Skip cancelled predecessors
            //从队列中删除当前节点之前状态为CANCELLED的节点
            Node pred = node.prev;
            while (pred.waitStatus > 0)
                node.prev = pred = pred.prev;
    
            // predNext is the apparent node to unsplice. CASes below will
            // fail if not, in which case, we lost race vs another cancel
            // or signal, so no further action is necessary.
            Node predNext = pred.next;
    
            // Can use unconditional write instead of CAS here.
            // After this atomic step, other Nodes can skip past us.
            // Before, we are free of interference from other threads.
            //CANCELLED状态在这里设置
            node.waitStatus = Node.CANCELLED;
    
            // If we are the tail, remove ourselves.
            //如果当前是尾节点中断了,将其第一个非CANCELLED状态的前驱节点设置为新的尾节点,并设置pred.next=null
            //pred后面的节点将会被GC回收。注意,下面的两个原子操作,不管是否成功,都没有重试
            if (node == tail && compareAndSetTail(node, pred)) {
                compareAndSetNext(pred, predNext, null);
            } else {
                // If successor needs signal, try to set pred's next-link
                // so it will get one. Otherwise wake it up to propagate.
                int ws;
                if (pred != head &&
                    ((ws = pred.waitStatus) == Node.SIGNAL ||
                     (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                    pred.thread != null) {
                    Node next = node.next;
                    //当前线程对应的节点不是尾节点,其有后继节点并且后继节点不是CANCELLED状态,
                    //通过修改指针将当前线程节点从队列删除(设置pred.next = node.next)
                    if (next != null && next.waitStatus <= 0)
                        compareAndSetNext(pred, predNext, next);
                } else {
                	//根据前面的if条件,在以下几种情况时会执行到这里,唤醒node节点的后继节点
                	//1)pred=head,即当前被中断的线程前面的所有线程都是CANCELLED状态
                	//2)pred!=head,但是pred节点的状态不等于SIGNAL,且将pred节点的状态修改为SIGNAL失败
                	//3)pred节点记录的线程是null,目前已知头结点的thread字段确实为null,除此之外还有其他情况吗???
                    unparkSuccessor(node);
                }
    
                node.next = node; // help GC
            }
        }
    

    整个await()方法的流程到这里就分析完了,下面是归纳一下整个流程:

    • countDown
      现在,终于可以开始看看countDown方法的逻辑了:
        public void countDown() {
            sync.releaseShared(1);
        }
    
        public final boolean releaseShared(int arg) {
        	//之前分析过,该方法会将state的值减1,并且只有在减1后state=0才会返回true,表示计数器结束了
            if (tryReleaseShared(arg)) {
            	//唤醒后继节点中第一个不为CANCELLED状态的节点
                doReleaseShared();
                return true;
            }
            return false;
        }
    

    当一个线程将state修改成0时,顺便还要执行doReleaseShared方法,这个方法会将头结点的后继节点唤醒。
    有一个小细节需要注意,doReleaseShared方法在源码中有两个地方调用,一个入口就是刚讲的countDown方法,另一个就是从await方法进入,在setHeadAndPropagate中调用,但是二者是有先后顺序的是,是countDown方法唤醒最前面的线程之后,再由该线程依次唤醒后面的线程。

    5.流程梳理

    分析到这里,终于把CountDownLatch中的所有方法全部讲完了。但是仅仅分析代码仍然是不够的,因为本人分析到这里的时候,脑袋仍然是蒙的,主要原因是缺少一个全局的认识。代码放在这里都能看懂,但是代码为什么这样写?当计数器结束(即state=0)时,队列中的等待线程是一起全部换新,还是一个一个依次唤醒?线程被唤醒后重新执行doAcquireSharedInterruptibly中的自旋时,和第一次执行到底有哪些地方不一样呢?因此,有必要对以上的逻辑进行整体梳理。
    看完这部分源码之后,发现核心的逻辑都包含在doAcquireSharedInterruptibly中,现在是时候回过头来整理一下该方法的逻辑了。

    • 线程正常唤醒的情况
      假设现在有一个线程t1执行了await方法,由于等待队列还没初始化,因此先构造一个空的头节点,并且把t1构造成节点加到队列中,如下图:

      接着,在shouldParkAfterFailedAcquire方法中修改头结点的状态:

      现在又有新的t2线程执行了await,此时队列的结构将更新为下图:

      即每添加一个节点到等待队尾,就将其前驱节点的状态更新为Node.SIGNAL(即-1),然后所有的线程都阻塞在parkAndCheckInterrupt方法里。现在,计数器已经结束,最后一个执行countDown方法的线程顺带执行了doReleaseShared方法,将头结点的waitStatus更新成了0,如下图:

      继续向下执行到unparkSuccessor方法,唤醒线程t1t1parkAndCheckInterrupt方法中醒来,继续自旋。t1的前置节点就是头结点head,且state=0t1开始执行setHeadAndPropagate,将自己设置为头结点,并在setHead方法中将threadprev字段都设置为空,如下图:

      线程t1接着执行doReleaseShared方法,把头节点(此时t1就是头结点)状态更新为0,并唤醒t2,开始执行await之后的逻辑,如下图:

      唤醒t2后,t1退出await方法,此时队列如下:

      t2开始执行后,同样把自己设置为头结点,如下:

      在执行setHeadAndPropagate方法时,t2没有后继节点了,仍然会执行doReleaseShared方法,但是在doReleaseShared方法中,t2既是头结点又是尾节点,那就什么也不做,直接结束并退出await方法,此时队列里就只剩下一个头结点了。
    • 线程响应中断的情况
      上面讲解了线程正常被唤醒的逻辑,现在来看看如果等待队列中的线程被中断会有什么不同。
      假设等待队列中还是有t1t2两个节点,如下图:

      此时t1线程被中断,t1会执行cancelAcquire中的代码逻辑,将节点状态置为CANCELLED,节点中保存的线程置为null,并调用unparkSuccessor唤醒t2t1退出await逻辑,如下图:

      此时t2开始从parkAndCheckInterrupt醒来,由于t2节点前面是t1节点,非头结点,自旋一次直接进入shouldParkAfterFailedAcquire方法,将t1从队列中删掉,后面的逻辑就跟上面的情况一样了,如下图:

      从上面的分析可以看出,线程响应中断后,将节点状态标记成了CANCELLED,节点并没有删除,节点的删除操作是在后面线程唤醒之后操作的。

    6.更新日志

    • 3.4更新了线程中断的示意图
    • 3.5更新了await()方法流程图
  • 相关阅读:
    阶乘递归实现
    队列
    1+2+3+...+100用递归实现
    快速排序C语言实现
    js的onfocus,onblur事件
    CSP2021 游记 菜到离谱
    700题复习计划
    [传递闭包] P2881 [USACO07MAR]排名的牛Ranking the Cows
    【笔记】序列分块
    【题解】UVA10930 A-Sequence
  • 原文地址:https://www.cnblogs.com/NaLanZiYi-LinEr/p/12391903.html
Copyright © 2011-2022 走看看