本文分享自华为云社区《【高并发】AQS案例详解》，作者： 冰 河。
AQS 的全称为 AbstractQueuedSynchronizer，是在 J.U.C（java.util.concurrent）下子包中的类。
一、AQS 的设计如下
（1）使用 Node 实现 FIFO 队列，可以用于构建锁或者其他同步装置的基础框架。
（2）利用了一个 int 类型表示状态
在 AQS 类中，有一个叫做 state 的成员变量。
基于 AQS 有一个同步组件 ReentrantLock，在 ReentrantLock 中，state 表示获取锁的线程数。如果 state=0,则表示还没有线程获取锁；如果 state=1,则表示有线程获取了锁；如果 state>1，则表示重入锁的数量。
（3）使用方法是继承
设计上基于模板方法，使用时需要继承 AQS，并覆写其中的方法
（4）子类通过继承并通过实现它的方法管理其状态{acquire 和 release}的方法操纵状态
（5）可以同时实现排它锁和共享锁模式（独占、共享）
站在使用者的角度，AQS 的功能主要分为两类：独占模式和共享模式。它的所有子类中要么实现并使用了它的独占功能的 API，要么使用了共享锁的功能，而不会同时使用两套 API。即便是它最有名的子类——ReentrantReadWriteLock，也是通过两个内部类——ReadLock（读锁）和 WriteLock（写锁）两套 API 来实现的。
二、AQS 内部实现的大体思路
首先，AQS 内部维护了一个 CLH 队列来管理锁，线程会首先尝试获取锁，如果失败，就将当前线程以及等待等信息封装成一个 Node 节点，加入到同步队列 SyncQueue，接着会不断循环尝试获取锁，获取锁的条件是当前节点为 Head 的直接后继节点才会尝试获取锁，如果失败，就会阻塞自己，直到自己被唤醒。而持有锁的线程释放锁的时候，会唤醒队列中的后继线程。基于这些基础的设计和思路，JDK 提供了许多基于 AQS 的子类，比如：CountDownLatch、Semaphore、CyclicBarrier、ReentrantLock、Condition、FutureTask 等
三、AQS 同步组件
CountDownLatch：闭锁，通过一个计数，来保证线程是否一直阻塞
Semaphore：控制同一时间并发线程的数目
CyclicBarrier：与 CountDownLatch 类似，都能阻塞进程；
ReentrantLock：可重入锁
Condition: 在使用时需要 ReentrantLock
FutureTask:对比 Runnable 和 Callable
1.CountDownLatch
同步辅助类，通过它可以阻塞当前线程。也就是说，能够实现一个线程或者多个线程一直等待，直到其他线程执行的操作完成。使用一个给定的计数器进行初始化，该计数器的操作是原子操作，即同时只能有一个线程操作该计数器。
调用该类 await()方法的线程会一直阻塞，直到其他线程调用该类的 countDown()方法，使当前计数器的值变为 0 为止。每次调用该类的 countDown()方法，当前计数器的值就会减 1。当计数器的值减为 0 的时候，所有因调用 await()方法而处于等待状态的线程就会继续往下执行。这种操作只能出现一次，因为该类中的计数器不能被重置。如果需要一个可以重置计数次数的版本，可以考虑使用 CyclicBarrier 类。
CountDownLatch 支持给定时间的等待，超过一定的时间不再等待，使用时只需要在 await()方法中传入需要等待的时间即可。此时，await()方法的方法签名如下：
public boolean await(long timeout, TimeUnit unit)
复制代码
CountDownLatch 使用场景
在某些业务场景中，程序执行需要等待某个条件完成后才能继续执行后续的操作。典型的应用为并行计算：当某个处理的运算量很大时，可以将该运算任务拆分成多个子任务，等待所有的子任务都完成之后，父任务再拿到所有子任务的运算结果进行汇总。
调用 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;@Slf4jpublic 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;@Slf4jpublic 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); }}
复制代码
2.Semaphore
控制同一时间并发线程的数目。能够完成对于信号量的控制，可以控制某个资源可被同时访问的个数。
提供了两个核心方法——acquire()方法和 release()方法。acquire()方法表示获取一个许可，如果没有则等待，release()方法则是在操作完成后释放对应的许可。Semaphore 维护了当前访问的个数，通过提供同步机制来控制同时访问的个数。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;@Slf4jpublic 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;@Slf4jpublic 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;@Slf4jpublic 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); }}
复制代码
3.CyclicBarrier
是一个同步辅助类，允许一组线程相互等待，直到到达某个公共的屏障点，通过它可以完成多个线程之间相互等待，只有当每个线程都准备就绪后，才能各自继续往下执行后面的操作。
与 CountDownLatch 有相似的地方，都是使用计数器实现，当某个线程调用了 CyclicBarrier 的 await()方法后，该线程就进入了等待状态，而且计数器执行加 1 操作，当计数器的值达到了设置的初始值，调用 await()方法进入等待状态的线程会被唤醒，继续执行各自后续的操作
CyclicBarrier 在释放等待线程后可以重用，所以，CyclicBarrier 又被称为循环屏障。
CyclicBarrier 使用场景
可以用于多线程计算数据，最后合并计算结果的场景
CyclicBarrier 与 CountDownLatch 的区别
（1）CountDownLatch 的计数器只能使用一次，而 CyclicBarrier 的计数器可以使用 reset()方法进行重置，并且可以循环使用
（2）CountDownLatch 主要实现 1 个或 n 个线程需要等待其他线程完成某项操作之后，才能继续往下执行，描述的是 1 个或 n 个线程等待其他线程的关系。而 CyclicBarrier 主要实现了多个线程之间相互等待，直到所有的线程都满足了条件之后，才能继续执行后续的操作，描述的是各个线程内部相互等待的关系。
（3）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;@Slf4jpublic 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.*;@Slf4jpublic 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;@Slf4jpublic 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); }}
复制代码
4.ReentrantLock 与锁
Java 中主要分为两类锁，一类是 synchronized 修饰的锁，另外一类就是 J.U.C 中提供的锁。
J.U.C 中提供的核心锁就是 ReentrantLock。
ReentrantLock（可重入锁）与 synchronized 区别：
（1）可重入性
二者都是同一个线程进入 1 次，锁的计数器就自增 1，需要等到锁的计数器下降为 0 时，才能释放锁。
（2）锁的实现
synchronized 是基于 JVM 实现的，而 ReentrantLock 是 JDK 实现的
（3）性能的区别
synchronized 优化之前性能比 ReentrantLock 差很多，但是自从 synchronized 引入了偏向锁，轻量级锁也就是自旋锁后，性能就差不多了。
（4）功能区别
ReentrantLock 独有的功能如下：
（1）ReentrantLock 可指定是公平锁还是非公平锁。而 synchronized 只能是非公平锁。所谓的公平锁就是先等待的线程先获得锁。
（2）提供了一个 Condition 类，可以分组唤醒需要唤醒的线程。而 synchronized 只能随机唤醒一个线程，或者唤醒全部的线程
（3）提供能够中断等待锁的线程的机制，lock.lockInterruptibly()。ReentrantLock 实现是一种自旋锁，通过循环调用 CAS 操作来实现加锁，性能上比较好是因为避免了使线程进入内核态的阻塞状态。
synchronized 能做的事情 ReentrantLock 都能做，而 ReentrantLock 有些能做的事情，synchronized 不能做。
在性能上，ReentrantLock 不会比 synchronized 差。
synchronized 的优势：
（1）不用手动释放锁，JVM 自动处理，如果出现异常，JVM 也会自动释放锁
（2）JVM 用 synchronized 进行管理锁定请求和释放时，JVM 在生成线程转储时能够锁定信息，这些对调试非常有价值，因为它们能标识死锁或者其他异常行为的来源。而 ReentrantLock 只是普通的类，JVM 不知道具体哪个线程拥有 lock 对象。
（3）synchronized 可以在所有 JVM 版本中工作，ReentrantLock 在某些 1.5 之前版本的 JVM 中可能不支持
ReentrantLock 中的部分方法说明：
boolean tryLock():仅在调用时锁定未被另一个线程保持的情况下才获取锁定
boolean tryLock(long, TimeUnit): 如果锁定在给定的等待时间内没有被另一个线程保持，且当前线程没有被中断，则获取这个锁定。
void lockInterruptibly():如果当前线程没有被中断，就获取锁定；如果被中断，就抛出异常
boolean isLocked():查询此锁定是否由任意线程保持
boolean isHeldByCurrentThread(): 查询当前线程是否保持锁定状态；
boolean isFair():判断是否是公平锁
boolean hasQueuedThread(Thread)：查询指定线程是否在等待获取此锁定
boolean hasQueuedThreads():查询是否有线程正在等待获取此锁定
boolean getHoldCount():查询当前线程保持锁定的个数
示例代码如下：
package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;@Slf4jpublic class LockExample { //请求总数 public static int clientTotal = 5000; //同时并发执行的线程数 public static int threadTotal = 200; public static int count = 0; private static final Lock lock = new ReentrantLock(); public static void main(String[] args) throws InterruptedException { ExecutorService executorService = Executors.newCachedThreadPool(); final Semaphore semaphore = new Semaphore(threadTotal); final CountDownLatch countDownLatch = new CountDownLatch(clientTotal); for(int i = 0; i < clientTotal; i++){ executorService.execute(() -> { try{ semaphore.acquire(); add(); semaphore.release(); }catch (Exception e){ log.error("exception", e); } countDownLatch.countDown(); }); } countDownLatch.await(); executorService.shutdown(); log.info("count:{}", count); } private static void add(){ lock.lock(); try{ count ++; }finally { lock.unlock(); } }}
复制代码
5.ReentrantReadWriteLock
在没有任何读写锁的时候，才可以取得写锁。如果一直有读锁存在，则无法执行写锁，这就会导致写锁饥饿。
示例代码如下：
package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;import java.util.Map;import java.util.Set;import java.util.TreeMap;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;import java.util.concurrent.locks.ReentrantReadWriteLock;@Slf4jpublic class LockExample {
private final Map<String, Data> map = new TreeMap<>(); private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); private final Lock readLock = lock.readLock(); private final Lock writeLock = lock.writeLock();
public Data get(String key){ readLock.lock(); try{ return map.get(key); }finally { readLock.unlock(); } }
public Set<String> getAllKeys(){ readLock.lock(); try{ return map.keySet(); }finally { readLock.unlock(); } }
public Data put(String key, Data value){ writeLock.lock(); try{ return map.put(key, value); }finally { writeLock.unlock(); } }
class Data{
}}
复制代码
6.StampedLock
控制锁三种模式：写、读、乐观读。
StampedLock 的状态由版本和模式两个部分组成，锁获取方法返回的是一个数字作为票据，用相应的锁状态来表示并控制相关的访问，数字 0 表示没有写锁被授权访问。
在读锁上分为悲观锁和乐观锁，乐观读就是在读操作很多，写操作很少的情况下，可以乐观的认为写入和读取同时发生的几率很小。因此，不悲观的使用完全的读取锁定。程序可以查看读取资料之后，是否遭到写入进行了变更，再采取后续的措施，这样的改进可以大幅度提升程序的吞吐量。
总之，在读线程越来越多的场景下，StampedLock 大幅度提升了程序的吞吐量。
StampedLock 源码中的案例如下，这里加上了注释
class Point { private double x, y; private final StampedLock sl = new StampedLock();
void move(double deltaX, double deltaY) { // an exclusively locked method long stamp = sl.writeLock(); try { x += deltaX; y += deltaY; } finally { sl.unlockWrite(stamp); } }
//下面看看乐观读锁案例 double distanceFromOrigin() { // A read-only method long stamp = sl.tryOptimisticRead(); //获得一个乐观读锁 double currentX = x, currentY = y; //将两个字段读入本地局部变量 if (!sl.validate(stamp)) { //检查发出乐观读锁后同时是否有其他写锁发生？ stamp = sl.readLock(); //如果没有，我们再次获得一个读悲观锁 try { currentX = x; // 将两个字段读入本地局部变量 currentY = y; // 将两个字段读入本地局部变量 } finally { sl.unlockRead(stamp); } } return Math.sqrt(currentX * currentX + currentY * currentY); }
//下面是悲观读锁案例 void moveIfAtOrigin(double newX, double newY) { // upgrade // Could instead start with optimistic, not read mode long stamp = sl.readLock(); try { while (x == 0.0 && y == 0.0) { //循环，检查当前状态是否符合 long ws = sl.tryConvertToWriteLock(stamp); //将读锁转为写锁 if (ws != 0L) { //这是确认转为写锁是否成功 stamp = ws; //如果成功 替换票据 x = newX; //进行状态改变 y = newY; //进行状态改变 break; } else { //如果不能成功转换为写锁 sl.unlockRead(stamp); //我们显式释放读锁 stamp = sl.writeLock(); //显式直接进行写锁 然后再通过循环再试 } } } finally { sl.unlock(stamp); //释放读锁或写锁 } }}
复制代码
示例代码如下：
package io.binghe.concurrency.example.lock;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;import java.util.concurrent.locks.StampedLock;@Slf4jpublic class LockExample { //请求总数 public static int clientTotal = 5000; //同时并发执行的线程数 public static int threadTotal = 200;
public static int count = 0;
private static final StampedLock lock = new StampedLock();
public static void main(String[] args) throws InterruptedException { ExecutorService executorService = Executors.newCachedThreadPool(); final Semaphore semaphore = new Semaphore(threadTotal); final CountDownLatch countDownLatch = new CountDownLatch(clientTotal); for(int i = 0; i < clientTotal; i++){ executorService.execute(() -> { try{ semaphore.acquire(); add(); semaphore.release(); }catch (Exception e){ log.error("exception", e); } countDownLatch.countDown(); }); } countDownLatch.await(); executorService.shutdown(); log.info("count:{}", count); }
private static void add(){ //加锁时返回一个long类型的票据 long stamp = lock.writeLock(); try{ count ++; }finally { //释放锁的时候带上加锁时返回的票据 lock.unlock(stamp); } }}
复制代码
总结：
（1）当只有少量竞争者时，synchronized 是一个很好的通用锁实现
（2）竞争者不少，但是线程的增长趋势是可预估的，此时，ReentrantLock 是一个很好的通用锁实现
（3）synchronized 不会引发死锁，其他的锁使用不当可能会引发死锁。
7.Condition
Condition 是一个多线程间协调通信的工具类，Condition 除了实现 wait 和 notify 的功能以外，它的好处在于一个 lock 可以创建多个 Condition，可以选择性的通知 wait 的线程
特点：
（1）Condition 的前提是 Lock，由 AQS 中 newCondition()方法 创建 Condition 的对象
（2）Condition await 方法表示线程从 AQS 中移除，并释放线程获取的锁，并进入 Condition 等待队列中等待，等待被 signal
（3）Condition signal 方法表示唤醒对应 Condition 等待队列中的线程节点，并加入 AQS 中，准备去获取锁。
示例代码如下：
package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.locks.Condition;import java.util.concurrent.locks.ReentrantLock;@Slf4jpublic class LockExample { public static void main(String[] args) { ReentrantLock reentrantLock = new ReentrantLock(); Condition condition = reentrantLock.newCondition();
new Thread(() -> { try { reentrantLock.lock(); log.info("wait signal"); // 1 condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } log.info("get signal"); // 4 reentrantLock.unlock(); }).start();
new Thread(() -> { reentrantLock.lock(); log.info("get lock"); // 2 try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } condition.signalAll(); log.info("send signal ~ "); // 3 reentrantLock.unlock(); }).start(); }}
复制代码
点击关注，第一时间了解华为云新鲜技术~
评论