并发编程 - 高性能队列 Disruptor

并发编程 - 深入了解 AbstractQueuedSynchronizer

在多线程编程中，锁是保证线程安全的核心机制之一。Java除了提供 `synchronized` 关键字外，还通过 java.util.concurrent.locks 包提供了更灵活的锁实现。`ReentrantLock` 是Java并发包中一个重要的锁实现，它提供了比 `synchronized` 更灵活的锁操作。接下来我们深入了解下Java中的可重入锁。

## ReentrantLock介绍

`ReentrantLock` 是Java并发包(java.util.concurrent.locks)中的一个可重入互斥锁实现。与`synchronized` 相比，它提供了以下增强功能：

* 可重入性

 ```
 同一线程可以多次获得同一把锁，注意避免死锁（例如递归调用场景）
 ```
* 公平性

 ```
 支持公平锁与非公平锁两种模式
 公平锁：按等待顺序分配锁（构造参数true）
 非公平锁：允许插队（默认模式，吞吐量更高）
 ```
* 锁等待可中断

 ```
 lockInterruptibly() 方法允许在等待锁时响应中断
 ```
* 超时获取锁

 ```
 tryLock() 方法可以指定超时时间
 ```
* 条件变量支持

 ```
 通过 newCondition() 创建多个条件队列
 ```

代码示例:

```
java

public class Counter {
 private final ReentrantLock lock = new ReentrantLock();
 private int count = 0;
 
 public void increment() {
 lock.lock(); // 阻塞获取锁
 try {
 count++;
 } finally {
 lock.unlock(); // 必须在finally中释放锁
 }
 }
}
```

关键点

锁释放必须放在finally中，防止异常导致死锁。

## ReentrantLock 锁的实现

基于 AQS 独占模式方式实现的互斥锁，其内部类分为 **公平锁**、**非公平锁**

### NonfairSync：非公平锁实现

非公平锁允许"插队"，当一个线程请求非公平锁时，如果在发出请求的同时该锁的状态变为可用，那么这个线程将跳过队列中所有等待的线程并获得这个锁，如果锁不可用，则与公平锁一样，进入队列排队等待。

对于非公平锁，只要 CAS 设置同步状态成功，则表示当前线程获取了锁

**构造函数**

`ReentrantLock` 对象默认构建时就是非公平锁

```
java

public ReentrantLock() {
 sync = new NonfairSync();
}
```

非公平锁要比公平锁性能高，减少了锁的上下文切换与调度时间，在没有必要严格执行线程顺序时，非公平锁是最好的选择，所以定义为默认

### FairSync：公平锁实现

公平的获取锁，也就是等待时间最长的线程最优先获取锁，也可以说锁获取是顺序的，根据 AQS 底层的 FIFO 的双向队列排队

**构造函数**

控制锁是否是公平锁

```
java

public ReentrantLock(boolean fair) {
 sync = fair ? new FairSync() : new NonfairSync();
}
```

## 实现原理

`ReentrantLock` 底层是基于 AQS 实现的，通过 AQS 维护的 `volatile` 的 `state` 变量来实现锁操作

对于锁的状态控制与可重入性实现，是通过AQS的 `state` 变量表示的，`state` 值为0时，表示锁未被占用，当 `state` 值不为0时，表示锁被占用，数值则是重入的次数。

等待队列是使用 CLH 变体实现线程排队，如果是公平锁模式，则严格按 FIFO 先进先出的顺序获取锁，如果是非公平锁模式，那每次新的线程会插队获取一次锁

可重入性

线程重复n次获取了锁，也就是重入了n次，随后在第n次释放该锁后，其他线程能够获取到该锁。锁的最终释放要求锁对于获取进行计数自增，计数表示当前锁被重复获取的次数，而锁被释放时，计数自减，当计数等于0时表示锁已经成功释放

## 源码分析

### lock 方法源码

`lock` 方法是如何实现让当前线程获取到锁资源（什么效果算是拿到了锁资源）

* 公平锁：调用 `acquire` 方法
* 非公平锁：上来先尝试将 `state` 从0修改为1，如果成功，代表获取锁资源。如果没有成功，调用 `acquire`

**公平&非公平的方法源码**

```
java

// 公平锁的sync的lock方法
final void lock() {
 acquire(1);
}

// 非公平锁的sync的lock方法
final void lock() {
 if (compareAndSetState(0, 1))
 setExclusiveOwnerThread(Thread.currentThread());
 else
 acquire(1);
}
```

**如果线程修改 state 失败怎么办？**

如果线程没有拿到锁资源，会到AQS的双向链表中排队等待（在期间，线程可能会挂起）

**AQS 的双向链表（队列）是个啥？**

AQS 中的双向链表是基于内部类 Node 在维护，Node中包含 `prev`，`next`，`thread` 属性，并且在 AQS 中还有两个属性，分别是 `head`，`tail`。

**AQS 的核心**

![2ccd93634b77487cafe88d7e8e809def.png](https://blog.liushigong.cn/static/img/5b64eb1310ea6e4aae04875480af64e0.2ccd93634b77487cafe88d7e8e809def.webp)

### acquire 方法源码

`acquire` 是一个业务方法，里面并没有实际的业务处理，都是在调用其他方法

```
java

// 核心acquire arg = 1
public final void acquire(int arg) {
 //1. 调用tryAcquire方法：尝试获取锁资源（非公平、公平），拿到锁资源，返回true，直接结束方法。 没有拿到锁资源，
 // 需要执行&&后面的方法

 //2. 当没有获取锁资源后，会先调用addWaiter：会将没有获取到锁资源的线程封装为Node对象，
 // 并且插入到AQS的队列的末尾，并且作为tail

 //3. 继续调用acquireQueued方法，查看当前排队的Node是否在队列的前面，如果在前面（head的next），尝试获取锁资源
 // 如果没在前面，尝试将线程挂起，阻塞起来！
 if (!tryAcquire(arg) &&
 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
 selfInterrupt();
}
```

### tryAcquire 方法源码

`tryAcquire` 分为公平和非公平两种实现，主要做了两件事：

* 如果state为0，尝试获取锁资源
* 如果state不为0，看一下是不是锁重入操作

#### 非公平锁

```
java

// 非公平锁实现！
final boolean nonfairTryAcquire(int acquires) {
 // 拿到当前线程！
 final Thread current = Thread.currentThread();
 // 拿到AQS的state
 int c = getState();
 // 如果state == 0，说明没有线程占用着当前的锁资源
 if (c == 0) {
 // 没人占用锁资源，我直接抢一波（不管有没有线程在排队）
 if (compareAndSetState(0, acquires)) {
 // 将当前占用这个互斥锁的线程属性设置为当前线程
 setExclusiveOwnerThread(current);
 // 返回true，拿锁成功
 return true;
 }
 }
 // 当前state != 0，说明有线程占用着锁资源
 // 判断拿着锁的线程是不是当前线程（锁重入）
 else if (current == getExclusiveOwnerThread()) {
 // 将state再次+1
 int nextc = c + acquires;
 // 锁重入是否超过最大限制
 // 01111111 11111111 11111111 11111111 + 1
 // 10000000 00000000 00000000 00000000
 // 抛出error
 if (nextc < 0) 
 throw new Error("Maximum lock count exceeded");
 // 将值设置给state
 setState(nextc);
 // 返回true，拿锁成功
 return true;
 }
 return false;
}
```

#### 公平锁

```
java

// 公平锁实现
protected final boolean tryAcquire(int acquires) {
 // 拿到当前线程！
 final Thread current = Thread.currentThread();
 // 拿到AQS的state
 int c = getState();
 // 阿巴阿巴~~~~
 if (c == 0) {
 // 判断是否有线程在排队，如果有线程排队，返回true，配上前面的!，那会直接不执行返回最外层的false
 if (!hasQueuedPredecessors() &&
 // 如果没有线程排队，直接CAS尝试获取锁资源
 compareAndSetState(0, acquires)) {
 setExclusiveOwnerThread(current);
 return true;
 }
 }
 else if (current == getExclusiveOwnerThread()) {
 int nextc = c + acquires;
 if (nextc < 0)
 throw new Error("Maximum lock count exceeded");
 setState(nextc);
 return true;
 }
 return false;
}
```

### addWaiter 方法源码

在获取锁资源失败后，需要将当前线程封装为 Node 对象，并且插入到 AQS 队列的末尾

```
java

// 将当前线程封装为Node对象，并且插入到AQS队列的末尾
private Node addWaiter(Node mode) {
 // 将当前线程封装为Node对象，mode为null，代表互斥锁
 Node node = new Node(Thread.currentThread(), mode);
 // pred是tail节点
 Node pred = tail;
 // 如果pred不为null，有线程正在排队
 if (pred != null) {
 // 将当前节点的prev，指定tail尾节点
 node.prev = pred;
 // 以CAS的方式，将当前节点变为tail节点
 if (compareAndSetTail(pred, node)) {
 // 之前的tail的next指向当前节点
 pred.next = node;
 return node;
 }
 }
 // 添加的流程为， 自己prev指向、tail指向自己、前节点next指向我
 // 如果上述方式，CAS操作失败，导致加入到AQS末尾失败，如果失败，就基于enq的方式添加到AQS队列
 enq(node);
 return node;
}

// enq，无论怎样都添加进入
private Node enq(final Node node) {
 for (;;) {
 // 拿到tail
 Node t = tail;
 // 如果tail为null，说明当前没有Node在队列中
 if (t == null) { 
 // 创建一个新的Node作为head，并且将tail和head指向一个Node
 if (compareAndSetHead(new Node()))
 tail = head;
 } else {
 // 和上述代码一致！
 node.prev = t;
 if (compareAndSetTail(t, node)) {
 t.next = node;
 return t;
 }
 }
 }
}
```

### acquireQueued 方法源码

acquireQueued 方法会查看当前排队的 Node 是否是 head 的 next，如果是，尝试获取锁资源，如果不是或者获取锁资源失败那么就尝试将当前 Node 的线程挂起（unsafe.park()）

在挂起线程前，需要确认当前节点的上一个节点的状态必须是小于等于0，

* 如果为1，代表是取消的节点，不能挂起
* 如果为-1，代表挂起当前线程
* 如果为-2，-3，需要将状态改为-1之后，才能挂起当前线程

```
java

// acquireQueued方法
// 查看当前排队的Node是否是head的next，
// 如果是，尝试获取锁资源，
// 如果不是或者获取锁资源失败那么就尝试将当前Node的线程挂起（unsafe.park()）
final boolean acquireQueued(final Node node, int arg) {
 // 标识。
 boolean failed = true;
 try {
 // 循环走起
 for (;;) {
 // 拿到上一个节点
 final Node p = node.predecessor();
 if (p == head && // 说明当前节点是head的next
 tryAcquire(arg)) { // 竞争锁资源，成功：true，失败：false
 // 进来说明拿到锁资源成功
 // 将当前节点置位head，thread和prev属性置位null
 setHead(node);
 // 帮助快速GC
 p.next = null; 
 // 设置获取锁资源成功
 failed = false;
 // 不管线程中断。
 return interrupted;
 }
 // 如果不是或者获取锁资源失败,尝试将线程挂起
 // 第一个事情，当前节点的上一个节点的状态正常！
 // 第二个事情，挂起线程
 if (shouldParkAfterFailedAcquire(p, node) &&
				// 通过LockSupport将当前线程挂起
 parkAndCheckInterrupt())
 }
 } finally {
 if (failed)
 cancelAcquire(node);
 }
}

// 确保上一个节点状态是正确的
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 // 拿到上一个节点的状态
 int ws = pred.waitStatus;
 // 如果上一个节点为 -1
 if (ws == Node.SIGNAL)
 // 返回true，挂起线程
 return true;
 // 如果上一个节点是取消状态
 if (ws > 0) {
 // 循环往前找，找到一个状态小于等于0的节点
 do {
 node.prev = pred = pred.prev;
 } while (pred.waitStatus > 0);
 pred.next = node;
 } else {
 // 将小于等于0的节点状态该为-1
 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
 }
 return false;
}
```

### unlock 方法源码

释放锁资源：

* 将state-1。
* 如果state减为0了，唤醒在队列中排队的Node。（一定唤醒离head最近的）

释放锁不分公平和非公平，就一个方法。

```
java

// 真正释放锁资源的方法
public final boolean release(int arg) {
 // 核心的释放锁资源方法
 if (tryRelease(arg)) {
 // 释放锁资源释放干净了。 （state == 0）
 Node h = head;
 // 如果头节点不为null，并且头节点的状态不为0，唤醒排队的线程
 if (h != null && h.waitStatus != 0)、
 // 唤醒线程
 unparkSuccessor(h);
 return true;
 }
 // 释放锁成功，但是state != 0
 return false;
}
// 核心的释放锁资源方法
protected final boolean tryRelease(int releases) {
 // 获取state - 1
 int c = getState() - releases;
 // 如果释放锁的线程不是占用锁的线程，抛异常
 if (Thread.currentThread() != getExclusiveOwnerThread())
 throw new IllegalMonitorStateException();
 // 是否成功的将锁资源释放利索 （state == 0）
 boolean free = false;
 if (c == 0) {
 // 锁资源释放干净。
 free = true;
 // 将占用锁资源的属性设置为null
 setExclusiveOwnerThread(null);
 }
 // 将state赋值
 setState(c);
 // 返回true，代表释放干净了
 return free;
}

// 唤醒节点
private void unparkSuccessor(Node node) {
 // 拿到头节点状态
 int ws = node.waitStatus;
 // 如果头节点状态小于0，换为0
 if (ws < 0)
 compareAndSetWaitStatus(node, ws, 0);
 // 拿到当前节点的next
 Node s = node.next;
 // 如果s == null ，或者s的状态为1
 if (s == null || s.waitStatus > 0) {
 // next节点不需要唤醒，需要唤醒next的next
 s = null;
 // 从尾部往前找，找到状态正常的节点。（小于等于0代表正常状态）
 for (Node t = tail; t != null && t != node; t = t.prev)
 if (t.waitStatus <= 0)
 s = t;
 }
 // 经过循环的获取，如果拿到状态正常的节点，并且不为null
 if (s != null)
 // 唤醒线程
 LockSupport.unpark(s.thread);
}
```

**为什么唤醒线程时从尾部往前找，而不是从前往后找？**

因为在 `addWaiter` 操作时，是先将当前 Node 的 prev 指针指向前面的节点，然后是将 tail 赋值给当前Node，最后才是能上一个节点的 next 指针，指向当前 Node。

如果从前往后，通过 next 去找，可能会丢失某个节点，导致这个节点不会被唤醒~ 如果从后往前找，肯定可以找到全部的节点。

## 注意事项

* 锁释放：必须在 finally 中调用 `unlock()`，否则可能导致死锁。
* 避免嵌套：过度嵌套锁会增加死锁风险，尽量降低锁粒度。
* 性能考量：非公平锁在竞争激烈时吞吐量更高，但可能引发 **线程饥饿**。
* 监控工具：使用 `getHoldCount()`、`isHeldByCurrentThread()` 等辅助调试。

什么是线程饥饿？

‌线程饥饿（Thread Starvation）‌是指某些线程因长时间无法获得所需的资源或执行机会，导致它们长时间处于等待状态的现象。这种现象通常是由于系统资源分配不公平或不合理造成的‌

**线程饥饿的原因**

‌1）不公平的资源分配策略‌：例如，在优先级调度中，如果高优先级的线程持续占用资源，低优先级的线程可能永远得不到执行机会‌

‌2）锁的非公平实现‌：如果使用非公平锁（如默认的std::mutex），某些线程可能长期被排在队列后面‌

‌3）资源占用时间过长‌：某些线程长时间持有资源，导致其他线程无法获取该资源‌

‌4）死循环或不释放资源‌：如果某个线程获取资源后进入死循环或迟迟不释放资源，其他线程会一直等待‌

## 与 synchronized 对比与选择

| 特性 | ReentrantLock | synchronized |
| --- | --- | --- |
| 锁获取方式 | 显式调用lock()/unlock() | 隐式（代码块/方法） |
| 公平性 | 支持公平/非公平模式 | 非公平 |
| 锁中断 | ✅ lockInterruptibly() | ❌ |
| 超时尝试 | ✅ tryLock(timeout) | ❌ |
| 条件变量 | ✅ 多Condition支持 | 单一wait()/notify() |
| 锁释放 | 必须手动unlock()（防死锁） | 自动释放 |

`ReentrantLock` 通过更灵活的API弥补了 `synchronized` 的局限性，尤其适合复杂并发场景。但其手动管理特性也要求开发者更谨慎地处理锁的获取与释放。

与 synchronized 进行选择

`ReentrantLock` 的危险性比 `synchronized` 要高，如果忘记 `finally` 块中 `unlock` 释放锁，会导致锁一直无法释放，阻塞后续线程

优先考虑 `synchronized`，只有当 `synchronized` 无法满足需求，需要高级功能（如公平性、条件变量）时才考虑使用 `ReentrantLock`，否则，还是应该优先使用 `synchronized`

并发编程 - ReentrantLock 源码解读

并发编程 - ForkJoinPool 深度源码解析

平时多线程开发一般就是使用 `Runnable，Callable，Thread，FutureTask，ThreadPoolExecutor` 这些内容，之前聊到 FutureTask 是同步非阻塞，将这些内容组合在一起去解决一些并发编程的问题时，很多时候不能去完成多线程的编排，这个时候就需要异步非阻塞的 `CompletableFuture` 来处理这些任务之间的逻辑关系了。

## CompletableFuture

CompletableFuture 是一个异步任务编排工具，底层是对 Future 的扩展和增强， `implements Future` 接口和 `CompletionStage`，通过 `CompletionStage` 接口实现任务链式调用。其本质就是 `Future` 的增强版，支持函数式编程和流式调用。

异步执行，默认线程池是 `ForkJoinPool.commonPool()`，可通过构造函数指定线程池

### 扩展 - 函数式接口

在正式开始之前，需要先掌握一些函数式编程的基础，需要了解以下函数式接口

| 接口 | 调用方法 | 响应 |
| --- | --- | --- |
| `Runnable` | run() | 无入参，没有返回值 |
| `Supplier<T>` | T get() | 无入参、有返回值 |
| `Function<T, R>` | R apply(T t); | 1个入参、有返回值 |
| `Consumer<T>` | void accept(T t) | 1个入参、没有返回值 |
| `BiFunction<T, U, R>` | R apply(T t, U u) | 2个入参、有返回值 |
| `BiConsumer<T, U>` | void accept(T t, U u) | 2个入参、没有返回值 |

## 实际应用

CompletableFuture 如果不提供线程池的话，默认使用的 `ForkJoinPool`，而 `ForkJoinPool` 内部是守护线程，如果main线程结束了，守护线程会跟着一起结束。

### 零依赖

零依赖（不依赖链式调用的任何任务）包含 `supplyAsync` 与 `runAsync`两个方法，核心是异步任务执行，使用 `Supplier supplier` 与 `Runnable` 作为入参，属于零依赖范畴，调用时即立刻执行

* `CompletableFuture supplyAsync(Supplier supplier)` 异步执行任务、有返回值
* `CompletableFuture supplyAsync(Supplier supplier, Executor executor)` 异步执行任务、自定义线程池、有返回值
* `CompletableFuture<Void> runAsync(Runnable runnable)` 异步执行任务、没有返回值`
* `CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)` 异步执行任务、自定义线程池、没有返回值
* `CompletableFuture completedFuture(U value)`
* `T join()` 完成时返回结果值，如果异常完成则抛出 CompletionException 异常。

### 一元依赖

一元依赖包含 `thenApply`、`thenAccept`、`thenRun`、`thenCompose`、`whenComplete`、`handle`、`exceptionally` 方法，其流程依赖于 **上一个异步任务执行完成后，用于执行一个新的异步任务**，依赖1个 `CompletableFuture`，也就是先后执行的关系。

* `thenApply` 把前面任务的执行结果，做为Function参数，并返回当前Function处理的结果
* `thenAccept` 把前面任务的执行结果，做为Function参数没有返回值
* `thenRun` 无视前面任务的结果，执行 Runnable无入参，无返回值
* `thenCompose` 把前面任务的执行结果，做为Function参数返回 CompletableFuture 类型对象
* `whenComplete` 以前面任务结果作为第一个参数，会出现的异常信息为第二个参数没有返回值
* `handle` 以前面任务结果作为第一个参数，会出现的异常信息为第二个参数有返回值
* `exceptionally` 前面任务执行正常则返回前面任务的结果如果异常不为 null 则返回 Function 执行的结果

开发时注意

需要结合状态机管理思想（未完成/完成/异常），通过 `handle()/exceptionally()` 方法正确处理异常

代码示例：

```
java

public static void main(String[] args) throws IOException {
 ExecutorService executor = Executors.newFixedThreadPool(10);

 /*CompletableFuture<String> taskA = CompletableFuture.supplyAsync(() -> {
 String id = UUID.randomUUID().toString();
 System.out.println("执行任务A：" + id);
 return id;
 });
 CompletableFuture<String> taskB = taskA.thenApply(result -> {
 System.out.println("任务B获取到任务A结果：" + result);
 result = result.replace("-", "");
 return result;
 });

 System.out.println("main线程拿到结果：" + taskB.join());*/

 CompletableFuture<String> taskB = CompletableFuture.supplyAsync(() -> {
 String id = UUID.randomUUID().toString();
 System.out.println("执行任务A：" + id + "," + Thread.currentThread().getName());
 return id;
 }).thenApplyAsync(result -> {
 System.out.println("任务B获取到任务A结果：" + result + "," + Thread.currentThread().getName());
 result = result.replace("-", "");
 return result;
 },executor);

 System.out.println("main线程拿到结果：" + taskB.join());
}
```

### 二元依赖

二元依赖包含 `thenCombine`、`thenAcceptBoth`、`runAfterBoth`、`applyToEither`、`acceptEither`、`runAfterEither` 方法，其流程依赖于 **上一个异步任务与一个其他任务执行完成后，用于执行一个新的异步任务**，依赖2个 `CompletableFuture` 调用链任务。

* `thenCombine` 以前面任务与other入参任务的结果做为参数计算合并任务，有返回值（返回计算后的结果）
* `thenAcceptBoth` 以前面任务与other入参任务的结果做为参数计算合并任务，没有返回值
* `runAfterBoth` 在前面任务与other任务之后运行 Runnable没有返回值
* `applyToEither` 前面任务与other任务，任意任务完成后作为入参运行有返回值
* `acceptEither` 前面任务与other任务，任意任务完成后作为入参运行没有返回值
* `runAfterEither` 在任一异步之后运行，执行 Runnable 没有入参没有返回值

例如：任务A和任务B并行执行，等到任务A和任务B全部执行完毕后，再执行任务C。

```
java

// api
CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

// demo
public static void main(String[] args) throws IOException {
 CompletableFuture<Integer> taskC = CompletableFuture.supplyAsync(() -> {
 System.out.println("任务A");
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return 78;
 }).thenCombine(CompletableFuture.supplyAsync(() -> {
 System.out.println("任务B");
 try {
 Thread.sleep(2000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return 66;
 }), (resultA, resultB) -> {
 System.out.println("任务C");
 int resultC = resultA + resultB;
 return resultC;
 });

 System.out.println(taskC.join());
 System.in.read();
}
```

### 多元依赖

多元依赖包含 `thenCombine`、`thenAcceptBoth`、`runAfterBoth`、`applyToEither`、`acceptEither`、`runAfterEither` 方法，其流程依赖于 **上一个异步任务与一个其他任务执行完成后，用于执行一个新的异步任务**，多元依赖依赖多个 `CompletableFuture` 调用链任务。

* `CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)` 阻塞等待所有给定的`CompletableFuture` 执行结束，当所有 `CompletableFuture` 任务完成时返回一个无返回值类型的`CompletableFuture`
* `CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)` 阻塞等待任意一个给定的`CompletableFuture` 对象执行结束当任一 `CompletableFuture` 任务完成时返回一个Object类型的`CompletableFuture`

场景对比

* 同步模型: CPU资源大量浪费在阻塞等待上，导致CPU资源利用率低
* 异步模型: Future用于表示异步计算的结果，只能通过阻塞或者轮询的方式获取结果
* CompletableFuture: 对Future进行了扩展，可以通过设置回调的方式处理计算结果，同时也支持组合操作，支持进一步的编排"

### Future 方法

* `boolean cancel(boolean mayInterruptIfRunning)` 根据 `mayInterruptIfRunning` 是否中断任务
* `boolean isCancelled()` 在正常完成之前，如果取消了此 `CompletableFuture`，则返回 true
* `boolean isDone()` 已完成返回 true
* `V get()` 等待计算完成，然后返回结果
* `V get(long timeout, TimeUnit unit)` 最大等待timeout时间计算完成，然后返回结果

## CompletableFuture 源码分析

`CompletableFuture` 的源码内容特别多。不需要把所有源码都看了，更多的是要掌握整个`CompletableFuture` 的源码执行流程，以及任务的执行时机。

从 `CompletableFuture` 中比较简单的方法作为分析的入口，从而掌握整体执行的流程。

### 当前任务执行方式

将任务和 `CompletableFuture` 封装到一起，再执行封住好的具体对象的 `run` 方法即可

```
java

// 提交任务到CompletableFuture
public static CompletableFuture<Void> runAsync(Runnable runnable) {
 // asyncPool：执行任务的线程池
 // runnable：具体任务。
 return asyncRunStage(asyncPool, runnable);
}

// 内部执行的方法
static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
 // 对任务做非空校验
 if (f == null) throw new NullPointerException();
 // 直接构建了CompletableFuture的对象，作为最后的返回结果
 CompletableFuture<Void> d = new CompletableFuture<Void>();
 // 将任务和CompletableFuture对象封装为了AsyncRun的对象
 // 将封装好的任务交给了线程池去执行
 e.execute(new AsyncRun(d, f));
 // 返回构建好的CompletableFuture
 return d;
}

// 封装任务的AsyncRun类信息
static final class AsyncRun extends ForkJoinTask<Void> implements Runnable, AsynchronousCompletionTask {
 // 声明存储CompletableFuture对象以及任务的成员变量
 CompletableFuture<Void> dep; 
 Runnable fn;

 // 将传入的属性赋值给成员变量
 AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
 this.dep = dep; 
 this.fn = fn;
 }
 // 当前对象作为任务提交给线程池之后，必然会执行当前方法
 public void run() {
 // 声明局部变量
 CompletableFuture<Void> d; Runnable f;
 // 将成员变量赋值给局部变量，并且做非空判断
 if ((d = dep) != null && (f = fn) != null) {
 // help GC，将成员变量置位null，只要当前任务结束后，成员变量也拿不到引用。
 dep = null; fn = null;
 // 先确认任务没有执行。
 if (d.result == null) {
 try {
 // 直接执行任务
 f.run();
 // 当前方法是针对Runnable任务的，不能将结果置位null
 // 要给没有返回结果的Runnable做一个返回结果
 d.completeNull();
 } catch (Throwable ex) {
 // 异常结束！
 d.completeThrowable(ex);
 }
 }
 d.postComplete();
 }
 }
}
```

### 任务编排的存储&执行方式

首先如果要在前继任务处理后，执行后置任务的话。

有两种情况：

* 前继任务如果没有执行完毕，后置任务需要先放在stack栈结构中存储
* 前继任务已经执行完毕了，后置任务就应该直接执行，不需要在往stack中存储了。

如果单独采用thenRun在一个任务后面指定多个后继任务，CompletableFuture无法保证具体的执行顺序，而影响执行顺序的是前继任务的执行时间，以及后置任务编排的时机。

### 任务编排流程

```
java

// 编排任务，前继任务搞定，后继任务再执行
public CompletableFuture<Void> thenRun(Runnable action) {
 // 执行了内部的uniRunStage方法，
 // null：线程池，现在没给。
 // action：具体要执行的任务
 return uniRunStage(null, action);
}

// 内部编排任务方法
private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
 // 后继任务不能为null，健壮性判断
 if (f == null) throw new NullPointerException();
 // 创建CompletableFuture对象d，与后继任务f绑定
 CompletableFuture<Void> d = new CompletableFuture<Void>();
 // 如果线程池不为null，代表异步执行，将任务压栈
 // 如果线程池是null，先基于uniRun尝试下，看任务能否执行
 if (e != null || !d.uniRun(this, f, null)) {
 // 如果传了线程池，这边需要走一下具体逻辑
 // e：线程池
 // d：后继任务的CompletableFuture
 // this：前继任务的CompletableFuture
 // f：后继任务
 UniRun<T> c = new UniRun<T>(e, d, this, f);
 // 将封装好的任务，push到stack栈结构
 // 只要前继任务没结束，这边就可以正常的将任务推到栈结构中
 // 放入栈中可能会失败
 push(c);
 // 无论压栈成功与否，都要尝试执行以下。
 c.tryFire(SYNC);
 }
 // 无论任务执行完毕与否，都要返回后继任务的CompletableFuture
 return d;
}
```

### 查看后置任务执行时机

任务在编排到前继任务时，因为前继任务已经结束了，这边后置任务会主动的执行

```
java

// 后置任务无论压栈成功与否，都需要执行tryFire方法
static final class UniRun<T> extends UniCompletion<T,Void> {

 Runnable fn;
 // executor：线程池
 // dep：后置任务的CompletableFuture
 // src：前继任务的CompletableFuture
 // fn：具体的任务
 UniRun(Executor executor, CompletableFuture<Void> dep,CompletableFuture<T> src, Runnable fn) {
 super(executor, dep, src); this.fn = fn;
 }

 final CompletableFuture<Void> tryFire(int mode) {
 // 声明局部变量
 CompletableFuture<Void> d; CompletableFuture<T> a;
 // 赋值局部变量
 // (d = dep) == null：赋值加健壮性校验
 if ((d = dep) == null ||
 // 调用uniRun。
 // a：前继任务的CompletableFuture
 // fn：后置任务
 // 第三个参数：传入的是this，是UniRun对象
 !d.uniRun(a = src, fn, mode > 0 ? null : this))
 // 进到这，说明前继任务没结束，等！
 return null;
 dep = null; src = null; fn = null;
 return d.postFire(a, mode);
 }
}

// 是否要主动执行任务
final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
 // 方法要么正常结束，要么异常结束
 Object r; Throwable x;
 // a == null：健壮性校验
 // (r = a.result) == null：判断前继任务结束了么？
 // f == null：健壮性校验
 if (a == null || (r = a.result) == null || f == null)
 // 到这代表任务没结束。
 return false;
 // 后置任务执行了没？ == null，代表没执行
 if (result == null) {
 // 如果前继任务的结果是异常结束。如果前继异常结束，直接告辞，封装异常结果
 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
 completeThrowable(x, r);
 else
 // 到这，前继任务正常结束，后置任务正常执行
 try {
 // 如果基于tryFire(SYNC)进来，这里的C不为null，执行c.claim
 // 如果是因为没有传递executor，c就是null，不会执行c.claim
 if (c != null && !c.claim())
 // 如果返回false，任务异步执行了，直接return false
 return false;
 // 如果claim没有基于线程池运行任务，那这里就是同步执行
 // 直接f.run了。
 f.run();
 // 封装Null结果
 completeNull();
 } catch (Throwable ex) {
 // 封装异常结果
 completeThrowable(ex);
 }
 }
 return true;
}

// 异步的线程池处理任务
final boolean claim() {
 Executor e = executor;
 if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
 // 只要有线程池对象，不为null
 if (e == null)
 return true;
 executor = null; // disable
 // 基于线程池的execute去执行任务
 e.execute(this);
 }
 return false;
}
```

前继任务执行完毕后，基于嵌套的方式执行后置。

```
java

// A：嵌套了B+C， B：嵌套了D+E
// 前继任务搞定，遍历stack执行后置任务
// A任务处理完，解决嵌套的B和C
final void postComplete() {
 // f：前继任务的CompletableFuture
 // h：存储后置任务的栈结构
 CompletableFuture<?> f = this; Completion h;
 // (h = f.stack) != null：赋值加健壮性判断，要确保栈中有数据
 while ((h = f.stack) != null ||
 // 循环一次后，对后续节点的赋值以及健壮性判断，要确保栈中有数据
 (f != this && (h = (f = this).stack) != null)) {
 // t：当前栈中任务的后续任务
 CompletableFuture<?> d; Completion t;
 // 拿到之前的栈顶h后，将栈顶换数据
 if (f.casStack(h, t = h.next)) {
 if (t != null) {
 if (f != this) {
 pushStack(h);
 continue;
 }
 h.next = null; // detach
 }
 // 执行tryFire方法，
 f = (d = h.tryFire(NESTED)) == null ? this : d;
 }
 }
}

// 回来了 NESTED == -1
final CompletableFuture<Void> tryFire(int mode) {
 CompletableFuture<Void> d; CompletableFuture<T> a;
 if ((d = dep) == null ||
 !d.uniRun(a = src, fn, mode > 0 ? null : this))
 return null;
 dep = null; src = null; fn = null;
 // 内部会执行postComplete，运行B内部嵌套的D和E
 return d.postFire(a, mode);
}
```

### 源码执行流程图

![image.png](https://blog.liushigong.cn/static/img/a8f5693ac8d7aa23d46f6bc137f29edb.image.webp)

## 总结

`CompletableFuture` 最重要的就是解决了异步回调的问题，优雅的组合多个异步任务，帮助我们简化回调逻辑，避免了“回调地狱”。其中需要使用自定义线程池避免 `ForkJoinPool` 通用资源的竞争，结合状态机管理（未完成/完成/异常），通过 `handle()/exceptionally()` 方法正确处理异常，同时 `CompletableFuture` 是基于 `Future` 的增强，避免 `get()` 的阻塞调用。