多线程的应用与原理分析8（countdownlatch）

￡神魔★判官ぃ 2022-04-23 07:42 169阅读 0赞

> **文章最前**： 我是Octopus，这个名字来源于我的中文名--章鱼；我热爱编程、热爱算法、热爱开源。所有源码在我的个人[github][] ；这博客是记录我学习的点点滴滴，如果您对 Python、Java、AI、算法有兴趣，可以关注我的动态，一起学习，共同进步。

**相关文章：**

1.  [多线程的应用与原理分析1][1]
2.  [多线程的应用与原理分析2（线程的状态）][2]
3.  [多线程的应用与原理分析3（原子性、可见性、有序性）][3]
4.  [多线程的应用与原理分析4（synchronized）][4_synchronized]
5.  [多线程的应用与原理分析5（ReentrantLock）][5_ReentrantLock]
6.  [多线程的应用与原理分析6（ReentrantLock）][6_ReentrantLock]
7.  [多线程的应用与原理分析7（Condition）][7_Condition]
8.  [多线程的应用与原理分析8（countdownlatch）][8_countdownlatch]
9.  [多线程的应用与原理分析9（原子操作）][9]
10. [多线程的应用与原理分析10（Semaphore）][10_Semaphore]
11. [多线程的应用与原理分析11（线程池）][11]

--------------------

**限制**

JUC中提供了几个比较常用的并发工具类，比如CountDownLatch、CyclicBarrier、Semaphore。  
**CountDownLatch**  
        countdownlatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完毕再执行。从命名可以解读到countdown是倒数的意思，类似于我们倒计时的概念。  
       countdownlatch提供了两个方法，一个是countDown，一个是await， countdownlatch初始化的时候需要传入一个整数，在这个整数倒数到0之前，调用了await方法的程序都必须要等待，然后通过countDown来倒数。  
**使用案例：**

package lock;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * @author zhangyu
     * @version V1.0
     * @ClassName: CountDownLatchDemo
     * @Description: 测试countdownlatch（类似于火箭发射）
     * @date 2019/1/25 10:14
     **/
    
    
    public class CountDownLatchDemo {
        public static void main(String[] args) throws InterruptedException {
            CountDownLatch countDownLatch = new CountDownLatch(3);
            new Thread(() -> {
                countDownLatch.countDown();
            }, "t1").start();
    
            new Thread(() -> {
                countDownLatch.countDown();
            }, "t2").start();
    
            new Thread(() -> {
                countDownLatch.countDown();
            }, "t3").start();
    
            countDownLatch.await();
            System.out.println("所有线程执行完毕！");
        }
    }

从代码的实现来看，有点类似join的功能，但是比join更加灵活。CountDownLatch构造函数会接收一个int类型的参数作为计数器的初始值，当调用CountDownLatch的countDown方法时，这个计数器就会减一。

通过await方法去阻塞去阻塞主流程  
                        ![20190125105209733.png][]  
**使用场景**  
1. 通过countdownlatch实现最大的并行请求，也就是可以让N个线程同时执行；

2. 比如应用程序启动之前，需要确保相应的服务已经启动，比如我们之前在讲zookeeper的时候，通过原生api连  
接的地方有用到countDownLatch  
**源码分析**  
         CountDownLatch类存在一个内部类Sync，上节课我们讲过，它是一个同步工具，一定继承了  
AbstractQueuedSynchronizer。很显然，CountDownLatch实际上是是使得线程阻塞了，既然涉及到阻塞，就一定涉及到AQS队列。

**await**  
await函数会使得当前线程在countdownlatch倒计时到0之前一直等待，除非线程别中断；从源码中可以得知await方法会转发到Sync的acquireSharedInterruptibly

方法  
public void await() throws InterruptedException \{ sync.acquireSharedInterruptibly(1); \}  
**acquireSharedInterruptibly**  
         这块代码主要是判断当前线程是否获取到了共享锁; AQS有两种锁类型，一种是共享锁，一种是独占锁，在这里用的是共享锁，为什么要用共享锁，因为CountDownLatch可以多个线程同时通过。

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {
         if (Thread.interrupted()) //判断线程是否中断
             throw new InterruptedException();
         if (tryAcquireShared(arg) < 0) //如果等于0则返回1，否则返回-1，返回-1表示需要阻塞
             doAcquireSharedInterruptibly(arg);
    }

在这里，state的意义是count，如果计数器为0，表示不需要阻塞，否则，只有在满足条件的情况下才会被唤醒；

**doAcquireSharedInterruptibly**  
获取共享锁

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
         final Node node = addWaiter(Node.SHARED); //创建一个共享模式的节点添加到队列中
         boolean failed = true;
         try {
             for (;;) { //自旋等待共享锁释放，也就是等待计数器等于0。
             final Node p = node.predecessor(); //获得当前节点的前一个节点
             if (p == head) {
             int r = tryAcquireShared(arg);//就判断尝试获取锁
             if (r >= 0) {//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);
       }
    }

**setHeadAndPropagate**  
                                ![20190125113916593.png][]

PROPAGATE：值为-3，表示releaseShared需要被传播给后续节点

private void setHeadAndPropagate(Node node, int propagate) {
         Node h = head; // 记录头节点
         setHead(node); //设置当前节点为头节点
         //前面传过来的propagate是1，所以会进入下面的代码
         if (propagate > 0 || h == null || h.waitStatus < 0 ||
         (h = head) == null || h.waitStatus < 0) {
           Node s = node.next; //获得当前节点的下一个节点，如果下一个节点是空表示当前节点为最后一个节
    点，或者下一个节点是share节点
           if (s == null || s.isShared())
            doReleaseShared(); //唤醒下一个共享节点
       }
    }

**doReleaseShared**  
释放共享锁，通知后面的节点

private void doReleaseShared() {
        for (;;) {
            Node h = head; //获得头节点
            if (h != null && h != tail) { //如果头节点不为空且不等于tail节点
               int ws = h.waitStatus;
               if (ws == Node.SIGNAL) { //头节点状态为SIGNAL，
                   if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) //修改当前头节点的状态为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;
     }
    }

**countdown**  
以共享模式释放锁，并且会调用tryReleaseShared函数，根据判断条件也可能会调用doReleaseShared函数

public final boolean releaseShared(int arg) {
         if (tryReleaseShared(arg)) { //如果为true，表示计数器已归0了
             doReleaseShared(); //唤醒处于阻塞的线程
             return true;
        }
        return false;
    }

**tryReleaseShared**  
        这里主要是对state做原子递减，其实就是我们构造的CountDownLatch的计数器，如果等于0返回true，否则返回false；

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;
         }
    }

[github]: https://github.com/zhangyu345293721
[1]: https://blog.csdn.net/zy345293721/article/details/86565631
[2]: https://blog.csdn.net/zy345293721/article/details/86583663
[3]: https://blog.csdn.net/zy345293721/article/details/86596268
[4_synchronized]: https://blog.csdn.net/zy345293721/article/details/86607632
[5_ReentrantLock]: https://blog.csdn.net/zy345293721/article/details/86617801
[6_ReentrantLock]: https://blog.csdn.net/zy345293721/article/details/86621392
[7_Condition]: https://blog.csdn.net/zy345293721/article/details/86622891
[8_countdownlatch]: https://blog.csdn.net/zy345293721/article/details/86636846
[9]: https://blog.csdn.net/zy345293721/article/details/86645245
[10_Semaphore]: https://blog.csdn.net/zy345293721/article/details/86644996
[11]: https://blog.csdn.net/zy345293721/article/details/86646019
[20190125105209733.png]: /images/20220323/a67ec427b65a4bc5964e205bf16401a3.png
[20190125113916593.png]: /images/20220323/934b3011c60544f1b3330796291304d4.png