亲宝软件园·资讯

展开

CountDownLatch源码探究 (JDK 1.8)

纳兰小依 人气:0

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;
                    }
                }
                //执行到这里说明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;
    }

之前对节点的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 (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;
        }
    }

头结点的状态成功更新为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后面的节点将会被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状态,通过修改指针将当前线程节点从队列删除
                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()的逻辑分析完,但是仅仅分析代码仍然是不够的,因为本人分析到这里的时候,脑袋仍然是蒙的,主要原因是缺少一个全局的认识。代码放在这里都能看懂,但是代码为什么这样写?当计数器结束(即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方法,此时队列里就只剩下一个头结点了。

  • 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方法唤醒最前面的线程之后,再由该线程依次唤醒后面的线程。

加载全部内容

相关教程
猜你喜欢
用户评论