JUC系列回顾之-CountDownLatch底层原理和示例

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号运动员:开跑...

总结:CountDownLatch通过AQS里面的共享锁来实现的,在创建CountDownLatch时候,会传递一个参数count,该参数是锁计数器的初始状态,表示该共享锁能够被count个线程同时获取。当某个线程调用CountDownLatch对象的await方法时候,该线程会等待共享锁可获取时,才能获取共享锁继续运行,而共享锁可获取的的条件是state == 0,而锁倒数计数器的初始值为count,每当一个线程调用该CountDownLatch对象的countDown()方法时候,计数器才-1,所以必须有count个线程调用该countDown()方法后,锁计数器才为0,这个时候等待的线程才能继续运行。




上一篇:MSSQL随机数概率测试


下一篇:设置Tomcat默认界面