CountDownLatch 是一个同步工具类，允许一个线程或者多个线程等待其他线程完成操作，再执行。
CountDownLatch(int count) 构造一个用给定计数初始化的 CountDownLatch。 // 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。 void await() // 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断或超出了指定的等待时间。 boolean await(long timeout, TimeUnit unit) // 递减锁存器的计数，如果计数到达零，则释放所有等待的线程。 void countDown() // 返回当前计数。 long getCount() // 返回标识此锁存器及其状态的字符串。 String toString()
CountDownLatch和CyclicBarrier的区别：
(1).CountDownLatch 的作用是允许1或者多个线程，等待另外N个线程完成某件事情之后，这1个或者多个线程才能执行。CyclicBarrier 是N个线程相互等待，任何一个线程完成任务之前，所有的线程必须等待。
(2).CountDownLatch 计数器是一次性的，无法被重置的，而CyclicBarrier的计数器在调用reset方法之后，还可以重新使用，因此被称为循环的barrier。
CountDownLatch 底层实现：
1.构造方法：创建一个Sync对象，而Sync继承AQS。
/**
* Constructs a {@code CountDownLatch} initialized with the given count.
*
* @param count the number of times {@link #countDown} must be invoked
* before threads can pass through {@link #await}
* @throws IllegalArgumentException if {@code count} is negative
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
2.Sync 是CountDownLatch的内部私有类，组合到CountDownLatch里：
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
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;
}
}
}
private final Sync sync;
在AQS中state是一个private volatile int类型的对象。CountDownLatch使用state来计数，CountDownLatch的getCount最终调用的是AQS的getState()
,返回state进行计数。
3.await()方法：调用AQS的acquireSharedInterruptibly方法
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
//1.获取共享锁
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//判断线程是否为中断状态，如果是抛出interruptedException
if (Thread.interrupted())
throw new InterruptedException();
//尝试获取共享锁，尝试成功就返回，否则调用doAcquireSharedInterruptibly方法
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
//2.尝试获取共享锁，重写AQS里面的方法
protected int tryAcquireShared(int acquires) {
//锁状态 == 0，表示所没有被任何线程所获取，即是可获取的状态，否则锁是不可获取的状态
return (getState() == 0) ? 1 : -1;
}
//3.doAcquireSharedInterruptibly方法会使得当前线程一直等待，直到当前线程获取到锁（或被中断）才返回
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//创建“当前线程”的Node节点，且node中记录的锁是“共享锁”类型，并将节点添加到CLH队列末尾。
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
//获取前继节点，如果前继节点是等待锁队列的表头，则尝试获取共享锁
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (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);
}
}
/*说明：4.shouldParkAfterFailedAcquire 返回当前线程是否应该阻塞
(01) 关于waitStatus请参考下表(中扩号内为waitStatus的值)，更多关于waitStatus的内容，可以参考前面的Node类的介绍。
CANCELLED[1] -- 当前线程已被取消
SIGNAL[-1] -- “当前线程的后继线程需要被unpark(唤醒)”。一般发生情况是：当前线程的后继线程处于阻塞状态，而当前线程被release或cancel掉，因此需要唤醒当前线程的后继线程。
CONDITION[-2] -- 当前线程(处在Condition休眠状态)在等待Condition唤醒
PROPAGATE[-3] -- (共享锁)其它线程获取到“共享锁”
[0] -- 当前线程不属于上面的任何一种状态。
(02) shouldParkAfterFailedAcquire()通过以下规则，判断“当前线程”是否需要被阻塞。
规则1：如果前继节点状态为SIGNAL，表明当前节点需要被unpark(唤醒)，此时则返回true。
规则2：如果前继节点状态为CANCELLED(ws>0)，说明前继节点已经被取消，则通过先前回溯找到一个有效(非CANCELLED状态)的节点，并返回false。
规则3：如果前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL，并返回false。
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 前驱节点的状态
int ws = pred.waitStatus;
// 如果前驱节点是SIGNAL状态，则意味着当前线程需要unpark唤醒，此时返回true
if (ws == Node.SIGNAL)
return true;
// 如果前继节点是取消的状态，则设置当前节点的“当前前继节点为”原节点的前继节点
if (ws > 0) {
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.
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
4. countDown()源码 :
//1.该方法其实调用AQS中的releaseShared(1)释放共享锁方法。
public void countDown() { sync.releaseShared(1); }
//2.目的是让当前线程释放它所持有的共享锁，它首先会通过tryReleaseShared()去尝试释放共享锁。尝试成功，则直接返回；尝试失败，则通过doReleaseShared()去释放共享锁。
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
//3.tryReleaseShared()在CountDownLatch.java中被重写，释放共享锁，将锁计数器-1
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
// 获取“锁计数器”的状态
int c = getState();
if (c == 0)
return false;
// “锁计数器”-1
int nextc = c-1;
// 通过CAS函数进行赋值。
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
实例：
public class CountDownLatchTest1 {
private static int SPORTSMAN_COUNT = 10;
private static final Random random = new Random();
// 用于判断发令之前运动员是否已经进入准备状态，需要等待10个运动员准备就绪，占有锁，等待10个运动员完成，释放锁。
private static CountDownLatch readyLatch = new CountDownLatch(SPORTSMAN_COUNT);
// 用于判断裁判是否已经发令，占有锁，等待裁判发令完成，释放锁
private static CountDownLatch startLatch = new CountDownLatch(1);
public static void main(String[] args) {
// 用于判断发令之前运动员是否已经进入准备状态，需要等待10个运动员准备就绪，占有锁，等待10个运动员完成，释放锁。
// CountDownLatch readyLatch = new CountDownLatch(SPORTSMAN_COUNT);
// 用于判断裁判是否已经发令，占有锁，等待裁判发令完成，释放锁
// CountDownLatch startLatch = new CountDownLatch(1);
// 启动10个线程，也就是10个运动员，做准备工作
for (int i = 0; i < SPORTSMAN_COUNT; i++) {
Thread t = new Thread(new MyTask((i + 1) + "号运动员", readyLatch, startLatch));
t.start();
}
// 当前运动员在其他运动员准备就绪前一直等待，也就是说等readyLatch倒数计数器为0之前一直等待
try {
readyLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 裁判发令，释放锁
startLatch.countDown();
System.out.println("裁判：所有运动员准备完毕，开始...");
}
static class MyTask implements Runnable {
private Lock lock = new ReentrantLock();
private CountDownLatch ready;
private CountDownLatch start;
private String name;
/**
*
* (构造方法)
*
* @param ready
* @param start
* @param name 运动员名称
*/
public MyTask(String name, CountDownLatch ready, CountDownLatch start) {
this.ready = ready;
this.start = start;
this.name = name;
}
@Override
public void run() {
lock.lock();
try {
// 1. 写运动员准备就绪的逻辑,准备readyTime秒
int readyTime = random.nextInt(1000);
System.out.println(name + ":我需要" + readyTime + "秒的时间准备。");
try {
Thread.sleep(readyTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "我已经准备完毕！");
// 释放锁readyLatch-1，表示一个运动员已经就绪
ready.countDown();
try {
// 等待裁判发开始命令
start.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "：开跑...");
} catch (Exception e) {
// TODO: handle exception
} finally {
lock.unlock();
}
}
}
}
运行结果：
1号运动员:我需要757秒的时间准备。 2号运动员:我需要9秒的时间准备。 3号运动员:我需要602秒的时间准备。 4号运动员:我需要232秒的时间准备。 5号运动员:我需要454秒的时间准备。 6号运动员:我需要440秒的时间准备。 7号运动员:我需要333秒的时间准备。 8号运动员:我需要406秒的时间准备。 9号运动员:我需要613秒的时间准备。 10号运动员:我需要121秒的时间准备。 2号运动员我已经准备完毕！ 10号运动员我已经准备完毕！ 4号运动员我已经准备完毕！ 7号运动员我已经准备完毕！ 8号运动员我已经准备完毕！ 6号运动员我已经准备完毕！ 5号运动员我已经准备完毕！ 3号运动员我已经准备完毕！ 9号运动员我已经准备完毕！ 1号运动员我已经准备完毕！ 裁判：所有运动员准备完毕，开始... 10号运动员：开跑... 8号运动员：开跑... 3号运动员：开跑... 1号运动员：开跑... 2号运动员：开跑... 9号运动员：开跑... 5号运动员：开跑... 6号运动员：开跑... 7号运动员：开跑... 4号运动员：开跑...
原文：https://www.cnblogs.com/hanease/p/14897668.html