以前文章

两个线程交替执行输出，一个数字1-10，一个字符a-e ,打印出来12a34b56c78d910e

ReentrantLock lock unLock 原理分析

Condition await signal 阻塞和唤醒 原理分析

CountDownLatch 原理分析

当我们需要实现并发请求，或者一个线程需要等待其他线程执行完成之后再执行时 ，我们可以使用 CountDownLatch 。

应用程序实例

public class CountDownLatchDemo extends Thread{

    static CountDownLatch countDownLatch = new CountDownLatch(2);
    public static void main(String[] args) {

        // 开启多个线程，并在run方法中进行线程等待，
        // 只有 countDown  归 0时才会进行执行，
        // 这种方式就是模拟并发执行
        for (int i = 0; i < 3; i++) {
            new CountDownLatchDemo().start();
        }
        // 当然 因为countDownLatch(2) ,所以 执行一次 countDown 还是不行的，
        // 真是情况下可以设置为1嘛  ，这里是例子合并成了一个
        countDownLatch.countDown();

        // 该线程是先执行业务逻辑， 最后在 countDown ， 计数归0 ，
        // 场景有点类似，在该线程中预热缓存 ，在上面多个线程中读取缓存
        new Thread(()->{
            System.out.println(Thread.currentThread().getName()+"\t"+"先执行我。。。。");
            countDownLatch.countDown();
        },"ThreadD").start();
    }

    @Override
    public void run() {
        try {
            // 在 countDown 计数器为归0 之前一直在这里阻塞
            countDownLatch.await();
            System.out.println(Thread.currentThread().getName()+
                               "\t"+System.currentTimeMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果

ThreadD 先执行我。。。。

ThreadA 1559284347551

ThreadB 1559284347551

ThreadC 1559284347551

类图关系

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

疑问：

1. await 等待的原理： 当 CountDownLatchDemo 多个线程同时启动时，执行到 run() 后为什么会处于等待状态？
2. countDown 唤醒等待的原理： 而 当 countDown 执行多次后 计数器归 0 后，是怎么通知 CountDownLatchDemo 线程的？

await

我们知道在初始化 CountDownLatch 的时候， 会传入一个整数，每次执行一次 countDown() ,计数器减一，当计算器为 0 时，处于等待状态的线程才会继续运行。计数器未归 0 的时候这块代码到底是做了什么呢？

// 根据所给的参数构造一个实例
public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    // volatile int state = count;
    this.sync = new Sync(count);
}

从上面的构造可以知道， 初始化 CountDownLatch 的时候需要传入一个整数，该值最终被保存到 state 中。是volatile 修饰的，可以保证有序性和可见性。

// 导致当前线程等待，直到锁存器倒计数到零，
// 除非线程是被中断
public void await() throws InterruptedException {
    // 获得共享锁
    sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 如果线程被中断了，则抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 尝试获得共享锁，
    // 如果 计数器为0 则返回1，否则返回-1
    if (tryAcquireShared(arg) < 0)
        // 表示当前计数器还未归0 ，需要去获得共享锁，
        // 将当前线程封装成node 节点添加到同步队列中，并再次获取锁
        // 如果计数器还未归0 则获取锁失败，
        // 并修改当前线程的上一个节点的waitStatus=SIGNAL(-1),
        // 并挂起当前线程
        doAcquireSharedInterruptibly(arg);
}
// 如果当前的 计数器 为0 返回1 否则返回-1
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 创建一个共享锁，封装当前线程的节点到同步队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 拿到当前节点的上一个节点
            final Node p = node.predecessor();
            // 如果该当前线程的上一个节点 是 head 节点，则尝试去获得锁
            if (p == head) {
                // r=1 表示计数器已归0
                // r=-1 表示计数器还未归0
                int r = tryAcquireShared(arg);
                // 如果此时 正好CountDown 次数 使计数器归 0
                // 因为是自旋，当线程挂起后再次被唤醒后还是会执行到这块代码
                
                // todo 这块等分析完 countDown 之后再来分析
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 获得锁失败后，
            // 将该节点的上一个节点 waitStatus设置为-1，
            // 删除取消的节点
            // 如果当前节点的上一个节点 的waitStatus=-1时，返回true
            
            // 这个方法在分析 lock 方法时已经分析过了，
            // 主要做了一下功能
            // 获得锁失败后是否应该挂起
    		// CANCELLED =  1 : 取消状态
    		// SIGNAL    = -1 ：只要前置节点释放锁，就会通知标识为 SIGNAL 状态的后续节点的线程
    		// CONDITION = -2 ：在Condition中（await,signal）中 会使用到
   			// PROPAGATE = -3 ：下一个acquireShared应该无条件传播
            // 1. 当 线程的pred线程被取消（1）时，会将该线程从竞争锁队列中删除
    		// 2. 当 线程的pred线程状态不是被取消（1)或者不是 -1 时，会将该线程的pred线程设置为 -1 :SIGNAL
    		// 3. 当 线程的pred线程状态是-1是，则返回true，表示可以进行挂起
            if (shouldParkAfterFailedAcquire(p, node) &&
                // park 当前线程
                // 如果返回true 表示已经中断
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

countDown

该方法主要是将计数器归0 ，说是释放锁，其实在 await 中并没有获得锁，所以这个 “释放锁” 只是在思想上与unlock 相似。

// 计数器递减，如果计数达到零释放所有等待的线程,
public void countDown() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    // 尝试释放共享锁
    // 实际上是将 计数器递减，如果递减后为0 ，则返回true
    if (tryReleaseShared(arg)) {
        // 如果在这里 计数器已经归0 了，便可以进行锁释放
        // 该方法主要是将 挂起的线程进行唤醒，修改waitStatus =-3 表示共享锁
        doReleaseShared();
        return true;
    }
    return false;
}
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;
        // 因为head 节点中并没有存储线程，
        // 如果head 和 tail 相同的话，说明同步队列中还没有等待获得锁的节点
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // head 的 waitStatus 如果 SIGNAL 表示可以去获取锁
            if (ws == Node.SIGNAL) {
                // 修改 head 的waitStatus =0 ，
                // 如果成功 唤醒head 的next 节点，
                // 如果失败 跳出当前循环：说明已经设置修改过了
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // unpark h.next
                // 唤醒 head节点的next节点
                // 这个方法在上面也已经分析过了，主要是唤醒当前节点的下一个节点
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     // 在上面if 的方法中，修改了head 的waitStatus =0 了，
                     // 这里在修改为 -3  ，表示是一个共享锁
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

await 被唤醒后的逻辑

从上面的 await 和 countDown 我们可以知道，根据实例化 CountDownLatch 时的构造参数的整数作为计数器。当执行 countDown 的时候计数器会递减，当计数器还未递减到0时，在await 方法中线程都会通过封装成一个node（thread=当前线程,waitStatus=0,nextWaiter=SHARED) 添加到同步队列中，当自旋获得不到锁时会将当前线程的上一个节点的waitStatus设置为-1，并挂起该线程。当 计数器递减为0 后， 会修改head节点的状态值为 waitStatus=PROPAGATE（-3），并唤醒head 节点的下一个节点。

此时我们接着分析，当挂起的线程被唤醒后又做了哪些事情。继续在 await 方法中的 doAcquireSharedInterruptibly（1） 来分析。

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 创建一个共享锁，封装当前线程的节点到同步队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 拿到当前节点的上一个节点
            final Node p = node.predecessor();
            // 如果该当前线程的上一个节点 是 head 节点，则尝试去获得锁
            if (p == head) {
                // r=1 表示计数器已归0
                // r=-1 表示计数器还未归0
                int r = tryAcquireShared(arg);
                // 如果此时 正好CountDown 次数 使计数器归 0
                // 因为是自旋，当线程挂起后再次被唤醒后还是会执行到这块代码
                if (r >= 0) {
                    // todo xxxx
                    // 设置当前节点为 head
                    setHeadAndPropagate(node, r);
                    // 并取消原head 的next 引用
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 获得锁失败后，
            // ..........
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
// 设置新的 head 节点，
// 此时是从 CountDownLatch 过来的，propagate 的值只能为 1 （即计数器归0）才会走到这里
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    // 设置新的 head 节点，
    setHead(node);
    /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
    // propagate > 0 表示当前计数器已归0
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        // 获得当前节点的下一个节点，如果是共享锁继续进行唤醒
        Node s = node.next;
        // 此时 CountDownLatch 过来的节点 是 nextWaiter = SHARED;
        // 所以是满足的，即共享锁，继续唤醒下一个节点
        if (s == null || s.isShared())
            // 此时head 节点已经更新了，
            // 所以继续执行该代码，会继续唤醒该节点的下个节点的线程
            doReleaseShared();
    }
}

疑问

1. await 等待的原理： 当 CountDownLatchDemo 多个线程同时启动时，执行到 run() 后为什么会处于等待状态？

答：当执行了 await 方法后，由于计数器不归0 ，尝试获得锁会失败。并将该线程添加到同步队列中（node=（thread=null,waitStatus=0,nextWaiter=SHARED）) 是一个标志共享锁的节点。并再次节点中获取锁，失败后修改该节点的上一个节点的状态 waitStatus=-1 ，并挂起当前线程。

2. countDown 唤醒等待的原理： 而 当 countDown 执行多次后 计数器归 0 后，是怎么通知 CountDownLatchDemo 线程的？

答：当执行完 await 后，由于计数器还未归0 ，所有线程当添加到同步队列中，当执行了 countDown 方法使计数器归0后会进行锁的释放。锁的释放主要是将head节点的 waitStatus=-3（共享） 并唤醒处于 head 节点的next 节点的线程(ThreadA)让其去竞争锁。被唤醒的线程再次回在 await 方法中获取锁，此时计数器归0 ，获得锁成功，该节点成为新的 head 。因为节点是共享锁，所以会再次调用锁的释放，再次去唤醒新 head 的next 节点，直至所有挂起的线程都获得锁。

1. 简介

简单描述CountDownLatch的功能，那就是

在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待

2. 实现原理

CountDownLatch 是通过一个计数器来实现的，当我们在 new 一个 CountDownLatch 对象的时候，需要带入该计数器值，该值就表示了线程的数量。

每当一个线程完成自己的任务后，计数器的值就会减 1 。当计数器的值变为0时，就表示所有的线程均已经完成了任务，然后就可以恢复等待的线程继续执行了。
如上图示例，用给定的计数为3进行初始化 CountDownLatch。由于调用了 countDown方法，所以在当前计数到达0之前，await方法会一直受阻塞；当计数到达0时会唤醒await方法阻塞的线程执行。

3. 源码结构

• J.U.C包中的最核心部分就是AQS的实现，它是JDK并发工具的实现基石，CountDownLatch也不例外，也是基于AQS进行实现
• await、countDown方法分别调用了AQS的doAcquireShared、doReleaseShared方法以共享锁的形式对AQS的共享变量state进行操作
• await方法阻塞等待，CountDownLatch 的作用是允许 1 或 N 个线程等待其他线程完成执行
• countDown方法变更计数，CountDownLatch 的计数器无法被重置

4. 实现剖析

4.1 await方法

• 调用await() 方法，来使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断
• 该方法内部使用 AQS 的 acquireSharedInterruptibly(int arg) 方法
• 在内部类 Sync 中重写了 tryAcquireShared(int arg) 方法
• 通过AQS中的getState() 方法，获取同步状态，其值等于计数器的值
• 如果计数器值不等于 0，则会调用 doAcquireSharedInterruptibly(int arg) 方法，该方法为一个自旋方法会尝试一直去获取同步状态

4.2 countDown方法

• 调用countDown() 方法，来改变计数器数量
• 内部调用AQS的releaseShared() 方法
• Sync重写了tryReleaseShared() 方法
• 释放锁，也就是操作计数器的过程，这里使用到了CAS（compareAndSetState）进行计数更新，若更新失败则进行自旋重试直到成功为止

4.3 getCount方法

调用AQS的getState方法获取计数

5. 实战用例

模拟主线程await阻塞，等待子线程计数countDown到0唤醒主线程执行

/**
 * created by guanjian on 2020/12/28 11:19
 */
public class CountDownLatchTest {

    private static final CountDownLatch cdl = new CountDownLatch(3);

    public static void main(String[] args) throws InterruptedException {
        System.out.println("开始");

        new Thread(() -> {
            try {
                System.out.format("当前计数=%s,需等待 \n", cdl.getCount());
                Thread.sleep(1000);
                cdl.countDown();
                System.out.format("当前计数=%s,需等待 \n", cdl.getCount());
                Thread.sleep(1000);
                cdl.countDown();
                System.out.format("当前计数=%s,需等待 \n", cdl.getCount());
                Thread.sleep(1000);
                cdl.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        cdl.await(6000, TimeUnit.MILLISECONDS);
        print();
        System.out.println("结束");
    }

    private static void print() {
        System.out.println("执行了");
    }
}

【控制台输出】
开始
当前计数=3,需等待 
当前计数=2,需等待 
当前计数=1,需等待 
执行了
结束

6. 总结

• CountDownLatch的作用实际是提供了一种多线程通信的方式
• AQS提供了多个线程并发且产生竞态条件时共享变量的安全读写方式，且丰富地提供了共享、排他两种处理竞态线程获取共享资源的方式
• CountDownLatch利用AQS的state共享变量作为信号量，await方法会在state不等于0时进行阻塞，countDown方法会操作state共享变量改变这个信号量的值从而来影响整个CountDownLatch的运行
• CountdownLatch是通过共享方式对锁进行操作，通过自旋方式进行阻塞等待，通过CAS对共享变量进行更新保证原子性

7. 参考

http://www.iocoder.cn/JUC/sike/CountDownLatch/

1.CountDownLatch工作原理

        CountDownLatch在多线程并发编程中充当一个计时器的功能，并且维护一个count的变量，并且其操作都是原子操作，该类主要通过countDown()和await()两个方法实现功能的，首先通过建立CountDownLatch对象，并且传入参数即为count初始值。如果一个线程调用了await()方法，那么这个线程便进入阻塞状态，并进入阻塞队列。如果一个线程调用了countDown()方法，则会使count-1；当count的值为0时，这时候阻塞队列中调用await()方法的线程便会逐个被唤醒，从而进入后续的操作。比如下面的例子就是有两个操作，一个是读操作一个是写操作，现在规定必须进行完写操作才能进行读操作。所以当最开始调用读操作时，需要用await()方法使其阻塞，当写操作结束时，则需要使count等于0。因此count的初始值可以定为写操作的记录数，这样便可以使得进行完写操作，然后进行读操作。

1. 首先是创建实例 CountDownLatch countDown = new CountDownLatch(2)
2. 需要同步的线程执行完之后，计数-1； countDown.countDown()
3. 需要等待其他线程执行完毕之后，再运行的线程，调用 countDown.await()实现阻塞同步

2. 应用场景

前面给了一个demo演示如何用，那这个东西在实际的业务场景中是否会用到呢？

因为确实在一个业务场景中使用到了，不然也就不会单独捞出这一节...

电商的详情页，由众多的数据拼装组成，如可以分成一下几个模块

• 交易的收发货地址，销量
• 商品的基本信息（标题，图文详情之类的）
• 推荐的商品列表
• 评价的内容
• ....

上面的几个模块信息，都是从不同的服务获取信息，且彼此没啥关联；所以为了提高响应，完全可以做成并发获取数据，如

• 线程1获取交易相关数据
• 线程2获取商品基本信息
• 线程3获取推荐的信息
• 线程4获取评价信息
• ....

但是最终拼装数据并返回给前端，需要等到上面的所有信息都获取完毕之后，才能返回，这个场景就非常的适合 CountDownLatch来做了

1. 在拼装完整数据的线程中调用 CountDownLatch#await(long, TimeUnit) 等待所有的模块信息返回
2. 每个模块信息的获取，由一个独立的线程执行；执行完毕之后调用 CountDownLatch#countDown() 进行计数-1
    

3.代码演示

package cn.day13;

import java.util.concurrent.CountDownLatch;

public class Test {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		final CountDownLatch latch = new CountDownLatch(2);

		new Thread() {
			public void run() {
				try {
					System.out.println("子线程" + Thread.currentThread().getName()
							+ "正在执行");
					Thread.sleep(3000);
					System.out.println("子线程" + Thread.currentThread().getName()
							+ "执行完毕");
					latch.countDown();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			};
		}.start();

		new Thread() {
			public void run() {
				try {
					System.out.println("子线程" + Thread.currentThread().getName()
							+ "正在执行");
					Thread.sleep(3000);
					System.out.println("子线程" + Thread.currentThread().getName()
							+ "执行完毕");
					latch.countDown();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			};
		}.start();

		try {
			System.out.println("等待2个子线程执行完毕...");
			latch.await();
			System.out.println("2个子线程已经执行完毕");
			System.out.println("继续执行主线程");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

运行结果：

子线程Thread-0正在执行
等待2个子线程执行完毕...
子线程Thread-1正在执行
子线程Thread-0执行完毕
子线程Thread-1执行完毕
2个子线程已经执行完毕
继续执行主线程

代码二

public class CountDownLatchDemo {
    private CountDownLatch countDownLatch;

    private int start = 10;
    private int mid = 100;
    private int end = 200;

    private volatile int tmpRes1, tmpRes2;

    private int add(int start, int end) {
        int sum = 0;
        for (int i = start; i <= end; i++) {
            sum += i;
        }
        return sum;
    }


    private int sum(int a, int b) {
        return a + b;
    }

    public void calculate() {
        countDownLatch = new CountDownLatch(2);

        Thread thread1 = new Thread(() -> {
            try {
                // 确保线程3先与1，2执行，由于countDownLatch计数不为0而阻塞
                Thread.sleep(100);
                System.out.println(Thread.currentThread().getName() + " : 开始执行");
                tmpRes1 = add(start, mid);
                System.out.println(Thread.currentThread().getName() +
                        " : calculate ans: " + tmpRes1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        }, "线程1");

        Thread thread2 = new Thread(() -> {
            try {
                // 确保线程3先与1，2执行，由于countDownLatch计数不为0而阻塞
                Thread.sleep(100);
                System.out.println(Thread.currentThread().getName() + " : 开始执行");
                tmpRes2 = add(mid + 1, end);
                System.out.println(Thread.currentThread().getName() +
                        " : calculate ans: " + tmpRes2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        }, "线程2");


        Thread thread3 = new Thread(()-> {
            try {
                System.out.println(Thread.currentThread().getName() + " : 开始执行");
                countDownLatch.await();
                int ans = sum(tmpRes1, tmpRes2);
                System.out.println(Thread.currentThread().getName() +
                        " : calculate ans: " + ans);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "线程3");

        thread3.start();
        thread1.start();
        thread2.start();
    }


    public static void main(String[] args) throws InterruptedException {
        CountDownLatchDemo demo = new CountDownLatchDemo();
        demo.calculate();

        Thread.sleep(1000);
    }
}

运行结果

线程3 : 开始执行
线程1 : 开始执行
线程2 : 开始执行
线程1 : calculate ans: 5005
线程2 : calculate ans: 15050
线程3 : calculate ans: 20055