countDownLatch
-
疯狂创客圈 经典图书 ： 《Netty Zookeeper Redis 高并发实战》 面试必备 + 面试必备 + 面试必备 【博客园总入口 】
-
疯狂创客圈 经典图书 ： 《SpringCloud、Nginx高并发核心编程》 大厂必备 + 大厂必备 + 大厂必备 【博客园总入口 】
-
入大厂+涨工资必备： 高并发【 亿级流量IM实战】 实战系列 【 SpringCloud Nginx秒杀】 实战系列 【博客园总入口 】
JUC 高并发工具类（3文章）与高并发容器类（N文章） ：
- 1 CyclicBarrier 使用&核心原理 图解
- 2 countDownLatch 使用&核心原理 图解
- 3 Semaphore 使用&核心原理 图解
- 4 跳表 核心原理 图解
- 5 ConcurrentSkipListMap – 秒懂
- 6 ConcurrentSkipListSet – 秒懂
1 CountDownLatch 是什么？
Java的concurrent包里面的CountDownLatch其实可以把它看作一个计数器，只不过这个计数器的操作是原子操作，同时只能有一个线程去操作这个计数器，也就是同时只能有一个线程去减这个计数器里面的值。
你可以向CountDownLatch对象设置一个初始的数字作为计数值，任何调用这个对象上的await()方法都会阻塞，直到这个计数器的计数值被其他的线程减为0为止。
CountDownLatch的一个非常典型的应用场景是：有一个任务想要往下执行，但必须要等到其他的任务执行完毕后才可以继续往下执行。假如我们这个想要继续往下执行的任务调用一个CountDownLatch对象的await()方法，其他的任务执行完自己的任务后调用同一个CountDownLatch对象上的countDown()方法，这个调用await()方法的任务将一直阻塞等待，直到这个CountDownLatch对象的计数值减到0为止。
比如：客户端一次请求5个统计数据，服务器需要全部统计完成后，才返回客户端，可以使用CountDownLatch 。
2 怎么使用 CyclicBarrier
2.1 构造方法
//参数count为计数值
public CountDownLatch(int count) { };
2.2 重要方法
//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
public void await() throws InterruptedException { };
//和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };
//将count值减1
public void countDown() { };
3 使用案例
使用步骤
- 首先是创建实例
CountDownLatch countDown = new CountDownLatch(2) - 需要同步的线程执行完之后，计数-1；
countDown.countDown() - 需要等待其他线程执行完毕之后，再运行的线程，调用
countDown.await()实现阻塞同步
示例代码
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个子线程已经执行完毕
继续执行主线程
4 CountDownLatch 使用场景
前面给了一个demo演示如何用，那这个东西在实际的业务场景中是否会用到呢？
因为确实在一个业务场景中使用到了，不然也就不会单独捞出这一节…
电商的详情页，由众多的数据拼装组成，如可以分成一下几个模块
- 交易的收发货地址，销量
- 商品的基本信息（标题，图文详情之类的）
- 推荐的商品列表
- 评价的内容
- ….
上面的几个模块信息，都是从不同的服务获取信息，且彼此没啥关联；所以为了提高响应，完全可以做成并发获取数据，如
- 线程1获取交易相关数据
- 线程2获取商品基本信息
- 线程3获取推荐的信息
- 线程4获取评价信息
- ….
但是最终拼装数据并返回给前端，需要等到上面的所有信息都获取完毕之后，才能返回，这个场景就非常的适合 CountDownLatch来做了
- 在拼装完整数据的线程中调用
CountDownLatch#await(long, TimeUnit)等待所有的模块信息返回 - 每个模块信息的获取，由一个独立的线程执行；执行完毕之后调用
CountDownLatch#countDown()进行计数-1
5 CountDownLatch 原理
CountDownLatch在多线程并发编程中充当一个计时器的功能，并且内部维护一个count的变量，并且其操作都是原子操作，该类主要通过countDown()和await()两个方法实现功能的，首先通过建立CountDownLatch对象，并且传入参数即为count初始值。
如果一个线程调用了await()方法，那么这个线程便进入阻塞状态，并进入阻塞队列。如果一个线程调用了countDown()方法，则会使count-1；当count的值为0时，这时候阻塞队列中调用await()方法的线程便会逐个被唤醒，从而进入后续的操作。比如下面的例子就是有两个操作，一个是读操作一个是写操作，现在规定必须进行完写操作才能进行读操作。所以当最开始调用读操作时，需要用await()方法使其阻塞，当写操作结束时，则需要使count等于0。因此count的初始值可以定为写操作的记录数，这样便可以使得进行完写操作，然后进行读操作。
构造方法：CountDownLatch
内部也是有个Sync类继承了AQS，所以CountDownLatch类的构造方法就是调用Sync类的构造方法，然后调用setState()方法设置AQS中state的值。
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
Sync(int count) {
setState(count);
}
方法：await()
该方法是使调用的线程阻塞住，直到state的值为0就放开所有阻塞的线程。实现会调用到AQS中的acquireSharedInterruptibly()方法，先判断下是否被中断，接着调用了tryAcquireShared()方法，讲AQS那篇文章里提到过这个方法是需要子类实现的，可以看到实现的逻辑就是判断state值是否为0，是就返回1，不是则返回-1。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
方法：countDown()
这个方法会对state值减1，会调用到AQS中releaseShared()方法，目的是为了调用doReleaseShared()方法，这个是AQS定义好的释放资源的方法，而tryReleaseShared()则是子类实现的，可以看到是一个自旋CAS操作，每次都获取state值，如果为0则直接返回，否则就执行减1的操作，失败了就重试，如果减完后值为0就表示要释放所有阻塞住的线程了，也就会执行到AQS中的doReleaseShared()方法。
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;
}
}
回到◀疯狂创客圈▶
疯狂创客圈 – Java高并发研习社群，为大家开启大厂之门