CountDownLatch内部机制

作者: 慧明小和尚下山去化斋 | 来源:发表于2017-02-20 19:59 被阅读0次

    前言

    在开始之前,就开始一个小实验,在这个实验里面,我希望main线程等待CountDownLatch减少它的计数器数值为0则结束,我规定这个数值为4,每隔一秒启动一个线程去CountDownLatch的计数器值。以下就是这个小实验的代码。

    public class CountDownLatchDemo {
        public static void main(String[] args) {
            //为了保持异步,首先启动一个线程
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < count; i++) {
                        try {
                            //每隔1秒启动一个线程减少计数器
                            Thread.sleep(1000L);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        new Thread(new Worker()).start();
                    }
                }
            }).start();
            System.out.println("等待CountDownLatch...");
            try {
                cdl.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("CountDownLatch结束...");
        }
    
        static final int count = 4;
        static final CountDownLatch cdl = new CountDownLatch(count);
    
        static class Worker implements Runnable {
    
            @Override
            public void run() {
                System.out.println(Thread.currentThread()+" count down");
                cdl.countDown();
            }
        }
    }
    

    这样,输出的结果也是意料之中的,如下所示:

    等待CountDownLatch...
    Thread[Thread-1,5,main] count down
    Thread[Thread-2,5,main] count down
    Thread[Thread-3,5,main] count down
    Thread[Thread-4,5,main] count down
    CountDownLatch结束...
    

    分析

    就像前面几篇分析的一样,其实都是利用了AQS提供的同步机制。现在就来看一看CountDownLatch的await方法。值得一提的是,CountDownLatch用AQS的state属性表示计数器的值。

        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
        //这个方法是在AQS里面定义的,CountDownLatch使用CLH队列的时候,它放进去的Node类型是Shared类型
        //与ReentrantLock不同,ReentrantLock使用的Exclusive类型。
        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            //如果线程设置了中断位,重新擦除中断标识以后抛出中断异常,这个方法也随之结束
            if (Thread.interrupted())
                throw new InterruptedException();
            
            // tryAcquireShared方法返回-1表示失败,如果失败的话,就要将线程“阻塞”
            // 这里的阻塞有两种方式,一种是通过死循环自旋,另一种是调用LockSupport的park方法将线程阻塞
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }
    

    那让调用await方法的线程停下,无法执行await以下的代码,这样的代码细节又是怎样的呢,再来看一看。

            protected int tryAcquireShared(int acquires) {
                //state属性=0,说明已经有线程把这个CountDownLatch的计数器操作为0
                return (getState() == 0) ? 1 : -1;
            }
    
            // 这个方法在countDown方法也会用到
            protected boolean tryReleaseShared(int releases) {
                // 把拿到的state属性减1(如果大于0),如果state为0,返回true
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    // 死循环的意义就在这里,如果CAS设置新的state失败,也就是说,线程之间的竞争很激烈,
                    // 它会一而再再而三的去尝试
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
            
        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) {
                        // 只有在CLH队头的线程才有资格去跳出这个方法
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
                            // 终于,计数器的值被置为0了
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    // state还不是0,就会走到这里,第一次的时候,waitStatus是0,那么node的waitStatus就会被置为SIGNAL
                    // 所以第二次走到这里的时候,parkAndCheckInterrupt方法就会被执行
                    // LockSupport的park方法就会把当前线程阻塞住。
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    

    当计数器为0的时候,不管调用await方法的线程是在自选,还是被LockSupport给阻塞住了,我们都得把它唤醒,唤醒这个线程的具体方法就在setHeadAndPropagate这个方法里边。

    setHeadAndPropagate

        private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; // Record old head for check below
            // 要唤醒等待的线程了,那么这个线程对应的node节点也就没有用了,我们就把它作为head指针。
            setHead(node);
            
            // 传进来的propagate一般都是1 > 0,或者原来的head没有被cancel掉,
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                // 如果没有后继的等待线程(也有可能因为null而不知道),
                // 或者都是Shared模式的节点,就全部释放这些线程。
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }
    

    doReleasedShared

        private void doReleaseShared() {
            for (;;) {
                Node h = head;
                // h = tail说明没有节点在CLH队列中,也就是没有线程在等待
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    // 在前面我们可以看到,head的waitStatus会被设置成SIGNAL然后被阻塞
                    // 如果是这样的话,就可以释放这个线程了
                    if (ws == Node.SIGNAL) {
                        // 在释放这个线程之前,应该先改变waitStatus,如果改变失败,再来一遍
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }
    

    unparkSuccessor

        private void unparkSuccessor(Node node) {
            
            // 把正在运行状态node设置成无用状态,也就是waitStatus设置为0
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            //找到下一个节点
            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                如果是null或者被cancel掉,就要从后向前找
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            //确实有这么一个后继节点,释放这个节点包含的线程
            //线程被唤醒之后,又回到上文写到的doAcquireSharedInterruptibly方法的死循环里
            if (s != null)
                LockSupport.unpark(s.thread);
        }
    

    countDown方法

    countDown方法也是调用AQS在CountDownLatch里的子类Sync实现的releaseShared方法。

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

    sync.releaseShared(1);

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

    没错,这里的tryReleaseShared方法和doReleaseShared方法就是上面写到的两个方法,有了上面的基础,理解countDown方法就不难了。首先,最后一个调用countDown方法的线程会先唤醒排在Sync队列队头的线程,然后被唤醒的线程就在doAcquireSharedInterruptibly方法里调用setHeadAndPropagate方法依次唤醒排在Sync队列的线程。

    结尾

    一个CountDownLatch调用了await方法,要么会被自旋,要么会被LockSupport给阻塞掉。如果是自旋的话,它会不断的去判断计数器是否已经被置0,置0了就可以在CLH队列设置新的head指针从而跳出await()方法。如果是被阻塞住了,那么最后一个调用countDown方法的线程就会负责去唤醒它,是通过调用LockSupport的unpark方法做到的。
    CountDownLatch是不能复用的,它和CyclicBarrier不一样,CyclicBarrier设置了一个Generation类作为复用的标识,同时还提供复用的机制,但是CountDownLatch并没有这么设计。

    相关文章

      网友评论

        本文标题:CountDownLatch内部机制

        本文链接:https://www.haomeiwen.com/subject/arwewttx.html