第12章 常用并发工具
4.1 CountDownLatch
4.1.1 CountDownLatch介绍
CountDownLatch是一种倒计时控制多线程的并发工具类，比如下边的例子：线程1需要在线程2和线程3完成工作后才可以继续执行，流程如下：
1、首先设定计数器（例子： new CountDownLatch(2)表示计数器为2）。
2、线程thread01调用awit()方法等待计算器清零。
3、线程thread02调用CountDownLatch的countDown()方法将计数器减1.
4、线程thread03调用CountDownLatch的countDown()方法将计数器减1.
5、计数器清零唤醒线程thread01继续执行。
CountDownLatch可以实现一个线程等多个线程、多个线程等一个线程，它的应用场景有很多，比如下边的社保查询例子：
分别用不同的线程查询居民信息和参保信息等信息，最后将每个信息进行汇总组装。
CountDownLatch的Api方法如下：
4.1.2 CountDownLatch测试
下边模拟两个线程查询参保状态、交费状态，另外一个线程进行信息组装，代码如下：
package com.yjoffer.javase.thread.waitnotify;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.xml.bind.helpers.ValidationEventImpl;
/**
- 并发工具测试
- @author 预见猿份(www.yjoffer.com)
-
*/
public class ThreadUtilsTest {
//测试CountDownLatch
public static void test_CountDownLatch() {
//计数器为2
CountDownLatch countDownLatch = new CountDownLatch(2);
ExecutorService threadPool = Executors.newCachedThreadPool();
threadPool.execute(()->{
logger.debug("组装线程等待中...");
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.debug("组装完成");
});
threadPool.execute(()->{
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.debug("参保状态");
//计数器减1
countDownLatch.countDown();
});
threadPool.execute(()->{
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.debug("交费状态");
//计数器减1
countDownLatch.countDown();
});
}
//休眠
public static void sleep(int time,TimeUnit timeUnit) {
try {
timeUnit.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
test_CountDownLatch1();
}
}输出：
21:52:43.491 FINE com.yjoffer.javase.thread2.waitnotify.ThreadUtilsTest[pool-1-thread-1] - 组装线程等待中...
21:52:44.492 FINE com.yjoffer.javase.thread2.waitnotify.ThreadUtilsTest[pool-1-thread-3] - 交费状态
21:52:45.491 FINE com.yjoffer.javase.thread2.waitnotify.ThreadUtilsTest[pool-1-thread-2] - 参保状态
21:52:45.492 FINE com.yjoffer.javase.thread2.waitnotify.ThreadUtilsTest[pool-1-thread-1] - 组装完成4.1.3 CountDownLatch工作原理
查看CountDownLatch类的源代码：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}sync类继承了AbstractQueuedSynchronizer（AQS类），源代码如下：
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;
}
}
}AQS是ReentrantLock、CountDownLatch等工具类的底层类，我们可以查一下ReentrantLock的源代码，如下：
AQS字面意思是“抽象队列同步器”，基于AQS自定义同步器可以实现独占锁、共享锁及并发工具。
AQS的工作原理的核心是共享状态state和先进先出双向队列，共享状态state作为多线程访问的共享变量，通过它可以控制锁的获取与释放，FIFO双向队列用于存放等待获取锁的线程，相当于线程的阻塞队列，下边通过一个同步锁的例子说明它的工作过程。如下图：
1、线程1获取锁，使用CAS将state由0更新为1成功，线程1即Node1作为队列的头结点。
2、线程2获取锁，使用CAS将state由0更新为1失败，线程2即Node2加入队尾等待锁的释放。
3、线程1释放锁，使用CAS将state由1更新为0成功，出队，唤醒它的后继结点Node2，线程2获取锁成功。
4、线程3去获取锁，加入队尾，等待线程2释放锁后被唤醒去获取锁。
CountDownLatch使用的是AQS的共享锁模式。
继续查看await()方法的源代码，最终找到Sync类的获取锁的方法：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}state状态为0表示锁空闲可获取锁，否则不允许获取，tryAcquireShared方法返回正数表示获取成功，负数表示失败。
在设定CountDownLatch计数器时指定count，源代码如下：
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}设定义的count即是state状态的初始值。
只有当state为0才表示计数清零，await线程才可以继续执行，否则进入阻塞队列阻塞。
如何让state清零呢？
其它线程执行countDown()方法表示state减1，源代码如下：
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return 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;
}
}在tryReleaseShared方法中找到答案，使用cas让state减1。
如果state等于0则tryReleaseShared方法返回true，唤醒等待的线程。
state等于0再次调用await()方法将不再阻塞。
请免费关注 预见猿份 公众号，发送：yjoffer.com，获取解锁密码
微信扫码关注关注公众号回复获取密码后，在下方输入并点击解锁，本机后续再读无需重复输入。