AQS的同步组件有哪些呢？

OSC开源社区 2023-03-16 484

描述

摘要：AQS 的全称为 Abstract Queued Synchronizer，是在 J.U.C（java.util.concurrent）下子包中的类。

AQS 的全称为 Abstract Queued Synchronizer，是在 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;
@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);
 }
}

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;
@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);
 }
}

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;
@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);
 }
}

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）功能区别

便利性：synchronized 使用起来比较方便，并且由编译器保证加锁和释放锁；ReentrantLock 需要手工声明加锁和释放锁，最好是在 finally 代码块中声明释放锁。

锁的灵活度和细粒度：在这点上 ReentrantLock 会优于 synchronized

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;
@Slf4j
public 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;
@Slf4j
public class LockExample {
 private final Map 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 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;
@Slf4j
public 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;
@Slf4j
public 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();
 }
}

审核编辑：刘清

打开APP阅读更多精彩内容