摘要：AQS 的全称为（AbstractQueuedSynchronizer） ，AQS 是一个用来构建锁和同步器的框架 ，使用 AQS 能简单且高效地构造出应用广泛的大量的同步器 。
本文分享自华为云社区《【高并发】AQS中的CountDownLatch 、Semaphore与CyclicBarrier核心用法》 ，作者： 冰 河 。
AQS 的全称为（AbstractQueuedSynchronizer） ，AQS 是一个用来构建锁和同步器的框架 ，使用 AQS 能简单且高效地构造出应用广泛的大量的同步器 。本文主要讲述AQS中的CountDownLatch 、Semaphore与CyclicBarrier核心用法 。
CountDownLatch
概述
同步辅助类 ，通过它可以阻塞当前线程 。也就是说 ，能够实现一个线程或者多个线程一直等待 ，直到其他线程执行的操作完成 。使用一个给定的计数器进行初始化 ，该计数器的操作是原子操作 ，即同时只能有一个线程操作该计数器 。
调用该类await()方法的线程会一直阻塞 ，直到其他线程调用该类的countDown()方法 ，使当前计数器的值变为0为止 。每次调用该类的countDown()方法 ，当前计数器的值就会减1 。当计数器的值减为0的时候 ，所有因调用await()方法而处于等待状态的线程就会继续往下执行 。这种操作只能出现一次 ，因为该类中的计数器不能被重置 。如果需要一个可以重置计数次数的版本 ，可以考虑使用CyclicBarrier类 。
CountDownLatch支持给定时间的等待 ，超过一定的时间不再等待 ，使用时只需要在await()方法中传入需要等待的时间即可 。此时 ，await()方法的方法签名如下：
使用场景
在某些业务场景中 ，程序执行需要等待某个条件完成后才能继续执行后续的操作 。典型的应用为并行计算：当某个处理的运算量很大时 ，可以将该运算任务拆分成多个子任务 ，等待所有的子任务都完成之后 ，父任务再拿到所有子任务的运算结果进行汇总 。
代码示例
调用ExecutorService类的shutdown()方法 ，并不会第一时间内把所有线程全部都销毁掉 ，而是让当前已有的线程全部执行完 ，之后 ，再把线程池销毁掉 。
示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await();
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
Thread.sleep(100);
log.info("{}", threadNum);
Thread.sleep(100);
}
}
支持给定时间等待的示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await(10, TimeUnit.MICROSECONDS);
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
Thread.sleep(100);
log.info("{}", threadNum);
}
}
Semaphore
概述
控制同一时间并发线程的数目 。能够完成对于信号量的控制 ，可以控制某个资源可被同时访问的个数 。
提供了两个核心方法——acquire()方法和release()方法 。acquire()方法表示获取一个许可 ，如果没有则等待 ，release()方法则是在操作完成后释放对应的许可 。Semaphore维护了当前访问的个数 ，通过提供同步机制来控制同时访问的个数 。Semaphore可以实现有限大小的链表 。
使用场景
Semaphore常用于仅能提供有限访问的资源 ，比如：数据库连接数 。
代码示例
每次获取并释放一个许可 ，示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
semaphore.acquire(); //获取一个许可
test(threadNum);
semaphore.release(); //释放一个许可
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
每次获取并释放多个许可 ，示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
semaphore.acquire(3); //获取多个许可
test(threadNum);
semaphore.release(3); //释放多个许可
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
假设有这样一个场景 ，并发太高了 ，即使使用Semaphore进行控制 ，处理起来也比较棘手 。假设系统当前允许的最高并发数是3 ，超过3后就需要丢弃 ，使用Semaphore也能实现这样的场景 ，示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
//尝试获取一个许可 ，也可以尝试获取多个许可 ，
//支持尝试获取许可超时设置 ，超时后不再等待后续线程的执行
//具体可以参见Semaphore的源码
if (semaphore.tryAcquire()) {
test(threadNum);
semaphore.release(); //释放一个许可
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
CyclicBarrier
概述
是一个同步辅助类 ，允许一组线程相互等待 ，直到到达某个公共的屏障点 ，通过它可以完成多个线程之间相互等待 ，只有当每个线程都准备就绪后 ，才能各自继续往下执行后面的操作 。
与CountDownLatch有相似的地方 ，都是使用计数器实现 ，当某个线程调用了CyclicBarrier的await()方法后 ，该线程就进入了等待状态 ，而且计数器执行加1操作 ，当计数器的值达到了设置的初始值 ，调用await()方法进入等待状态的线程会被唤醒 ，继续执行各自后续的操作 。CyclicBarrier在释放等待线程后可以重用 ，所以 ，CyclicBarrier又被称为循环屏障 。
使用场景
可以用于多线程计算数据 ，最后合并计算结果的场景
CyclicBarrier与CountDownLatch的区别
CountDownLatch的计数器只能使用一次 ，而CyclicBarrier的计数器可以使用reset()方法进行重置 ，并且可以循环使用
CountDownLatch主要实现1个或n个线程需要等待其他线程完成某项操作之后 ，才能继续往下执行 ，描述的是1个或n个线程等待其他线程的关系 。而CyclicBarrier主要实现了多个线程之间相互等待 ，直到所有的线程都满足了条件之后 ，才能继续执行后续的操作 ，描述的是各个线程内部相互等待的关系 。
CyclicBarrier能够处理更复杂的场景 ，如果计算发生错误 ，可以重置计数器让线程重新执行一次 。
CyclicBarrier中提供了很多有用的方法 ，比如：可以通过getNumberWaiting()方法获取阻塞的线程数量 ，通过isBroken()方法判断阻塞的线程是否被中断 。
代码示例
示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++){
final int threadNum = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
private static void race(int threadNum) throws Exception{
Thread.sleep(1000);
log.info("{} is ready", threadNum);
cyclicBarrier.await();
log.info("{} continue", threadNum);
}
}
设置等待超时示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++){
final int threadNum = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
private static void race(int threadNum) throws Exception{
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try{
cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
}catch (BrokenBarrierException | TimeoutException e){
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}
在声明CyclicBarrier的时候 ，还可以指定一个Runnable ，当线程达到屏障的时候 ，可以优先执行Runnable中的方法 。
示例代码如下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
log.info("callback is running");
});
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++){
final int threadNum = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
private static void race(int threadNum) throws Exception{
Thread.sleep(1000);
log.info("{} is ready", threadNum);
cyclicBarrier.await();
log.info("{} continue", threadNum);
}
}
点击关注 ，第一时间了解华为云新鲜技术~
声明：本站所有文章 ，如无特殊说明或标注 ，均为本站原创发布 。任何个人或组织 ，在未征得本站同意时 ，禁止复制 、盗用 、采集 、发布本站内容到任何网站 、书籍等各类媒体平台 。如若本站内容侵犯了原著者的合法权益 ，可联系我们进行处理 。
