平时多线程开发一般就是使用 `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 执行的结果


:::warning{title="开发时注意"}
需要结合状态机管理思想（未完成/完成/异常），通过 `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`

:::tip{title="场景对比"}

- 同步模型: 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()` 的阻塞调用。

并发编程 - CompletableFuture 任务编排与源码解读

并发编程 - FutureTask 应用&源码分析

Java 并发包（JUC）提供了一系列强大的并发工具类，本文将深入解析常用工具类的原理，并通过完整代码示例展示其实际应用场景




## 同步工具类

### CountDownLatch - 任务计数器

#### 原理

基于 AQS（AbstractQueuedSynchronizer）实现，初始化计数器值（state），线程通过countDown()释放共享锁递减计数state-1），调用await()的线程阻塞直到计数器归零（state=0）。

#### 本质

一次性协作机制，主线程等待指定数量的子任务完成。
 
#### 实现方法

CountDownLatch(int count)：构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N

- countDown()
 
 调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法会阻塞当前线程，直到N变成零
 由于countDown方法可以用在任何地方，所以这里说的N个 点，可以是N个线程，也可以是1个线程里的N个执行步骤
 在多个线程时，只需要把这个 CountDownLatch 的引用传递到线程里调用 countDown 方法即可
 
- await（long time，TimeUnit unit）：方法等待特定时间后，就会不再阻塞当前线程

#### 注意事项：

- 计数器不可重置（一次性）
- await()可设置超时时间避免永久阻塞
- 确保countDown()在finally块中调用
 
```java
// 模拟启动服务器（需所有服务就绪）
public class ServerInitializer {
 public static void main(String[] args) throws InterruptedException {
 int serviceCount = 3;
 CountDownLatch latch = new CountDownLatch(serviceCount);
 
 new Thread(new Service("Auth", 2000, latch)).start();
 new Thread(new Service("Database", 3000, latch)).start();
 new Thread(new Service("Cache", 1500, latch)).start();
 
 latch.await(); // 阻塞直到计数器归零
 System.out.println("所有服务启动完成！");
 }
 
 static class Service implements Runnable {
 private final String name;
 private final int bootTime;
 private final CountDownLatch latch;
 
 Service(String name, int bootTime, CountDownLatch latch) {
 this.name = name;
 this.bootTime = bootTime;
 this.latch = latch;
 }
 
 @Override
 public void run() {
 try {
 Thread.sleep(bootTime);
 System.out.println(name + " 服务已启动");
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 } finally {
 latch.countDown(); // 计数器减1
 }
 }
 }
}
```
 
 
### CyclicBarrier - 线程屏障
 
#### 原理

通过 `ReentrantLock` 和 `Condition` 实现，线程调用await()时计数减1并阻塞，当计数归零时唤醒所有线程并重置计数器。

#### 本质

可复用的同步点，用于多线程相互等待到达屏障点。

让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行

#### 注意事项：

- 支持通过Runnable定义屏障触发动作
- 使用reset()重置时需处理BrokenBarrierException
- 避免线程数大于parties导致永久阻塞


#### 实现方法
 
CyclicBarrier（int parties）：其参数表示屏障拦截的线程数量，每个线程调用 `await` 方法告诉`CyclicBarrier` 我已经到达了屏障，然后当前线程被阻塞

- reset()：重置计数器
- getNumberWaiting()：获取 CyclicBarrier 阻塞的数量
- isBroken()：阻塞的线程是否被中断
 
```java
// 多玩家游戏准备场景
public class GameSession {
 public static void main(String[] args) {
 int playerCount = 4;
 CyclicBarrier barrier = new CyclicBarrier(playerCount, 
 () -> System.out.println("所有玩家准备完成！开始游戏..."));
 
 for (int i = 1; i <= playerCount; i++) {
 new Thread(new Player("玩家" + i, barrier)).start();
 }
 }
 
 static class Player implements Runnable {
 private final String name;
 private final CyclicBarrier barrier;
 
 Player(String name, CyclicBarrier barrier) {
 this.name = name;
 this.barrier = barrier;
 }
 
 @Override
 public void run() {
 try {
 // 模拟加载资源时间
 int loadTime = ThreadLocalRandom.current().nextInt(1000, 3000);
 Thread.sleep(loadTime);
 System.out.println(name + " 准备完成 (" + loadTime + "ms)");
 
 barrier.await(); // 等待其他玩家
 
 // 游戏开始后的操作
 System.out.println(name + " 开始行动");
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 }
}
```

### Semaphore - 资源信号量

#### 原理

基于 AQS 实现，通过acquire()获取许可计数器减1（state-1），release()释放许可计数器加1（state+1），当计数器为0（state=0）时阻塞获取线程。

#### 本质

控制同时访问特定资源的线程数量，流量控制，限流

#### 注意事项：

- 公平模式（fair=true）避免线程饥饿

- 使用tryAcquire()尝试获取避免死锁

- 释放许可数量必须等于获取数量

 #### 实现方法
 
 Semaphore（int permits）： 构造函数定义许可证数量
 
 - tryAcquire()：尝试获取许可证
 - release()：归还许可证
 - intavailablePermits()：返回此信号量中当前可用的许可证数
 - intgetQueueLength()：返回正在等待获取许可证的线程数
 - booleanhasQueuedThreads()：是否有线程正在等待获取许可证
 - void reducePermits（int reduction）：减少reduction个许可证，是个protected方法
 - Collection getQueuedThreads()：返回所有等待获取许可证的线程集合，是个protected方 法

```java
// 数据库连接池模拟
public class ConnectionPool {
 private static final int POOL_SIZE = 5;
 private final Semaphore semaphore = new Semaphore(POOL_SIZE, true);
 private final Set<Connection> connections = new HashSet<>();
 
 public ConnectionPool() {
 for (int i = 0; i < POOL_SIZE; i++) {
 connections.add(createMockConnection());
 }
 }
 
 public Connection getConnection() throws InterruptedException {
 semaphore.acquire(); // 获取许可
 return getAvailableConnection();
 }
 
 public void releaseConnection(Connection conn) {
 if (conn != null) {
 returnConnection(conn);
 semaphore.release(); // 释放许可
 }
 }
 
 private synchronized Connection getAvailableConnection() {
 Connection conn = connections.iterator().next();
 connections.remove(conn);
 return conn;
 }
 
 private synchronized void returnConnection(Connection conn) {
 connections.add(conn);
 }
 
 private Connection createMockConnection() {
 return new Connection() {
 // 模拟连接实现
 };
 }
 
 // 使用示例
 public static void main(String[] args) {
 ConnectionPool pool = new ConnectionPool();
 ExecutorService executor = Executors.newFixedThreadPool(10);
 
 for (int i = 0; i < 10; i++) {
 executor.execute(() -> {
 try {
 Connection conn = pool.getConnection();
 System.out.println(Thread.currentThread().getName() 
 + " 获取连接，执行查询...");
 Thread.sleep(1000); // 模拟查询
 pool.releaseConnection(conn);
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 }
 });
 }
 
 executor.shutdown();
 }
}
```

## 高级同步工具类
 
### Exchanger - 数据交换器

Exchanger（交换者）是一个用于线程间协作的工具类
 
#### 原理
基于 `LockSupport.park/unpark`，每个交换点维护一个 "槽位"，首个到达线程存入数据并阻塞，第二个线程到达时交换数据

#### 本质

用于进行线程间的数据交换

它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据

这两个线程通过 exchange 方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也 执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方

#### 实现方法

- V exchange(V x)
- V exchange（V x，longtimeout，TimeUnit unit）设置最大等待时长


:::tip{title="应用"}
遗传算法：遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出2个交配结果

校对工作：采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对两个Excel数据进行校对，看看是否录入一致
:::
 
### Phaser - 灵活的阶段控制器

可重复使用的同步屏障，功能类似于 CyclicBarrier 和 CountDownLatch，但支持更灵活的使用。

#### 原理

基于树形结构减少竞争，每个阶段（phase）独立计数，支持动态注册/注销参与者，分层移除减少同步开销

#### 本质

可重用的同步屏障，支持动态调整参与者

```java
// 多阶段考试模拟
public class ExaminationSystem {
 public static void main(String[] args) {
 int studentCount = 5;
 Phaser phaser = new Phaser(1); // 注册主线程
 
 System.out.println("考试开始，学生入场");
 
 // 注册学生线程
 for (int i = 0; i < studentCount; i++) {
 phaser.register();
 new Thread(new Student(phaser), "学生" + (i+1)).start();
 }
 
 // 第一阶段：笔试
 phaser.arriveAndAwaitAdvance();
 System.out.println("\n笔试结束，开始机试");
 
 // 第二阶段：机试
 phaser.arriveAndAwaitAdvance();
 System.out.println("\n机试结束，开始面试");
 
 // 第三阶段：面试
 phaser.arriveAndAwaitAdvance();
 System.out.println("\n所有考试结束");
 phaser.arriveAndDeregister(); // 主线程退出
 }
 
 static class Student implements Runnable {
 private final Phaser phaser;
 
 Student(Phaser phaser) {
 this.phaser = phaser;
 }
 
 @Override
 public void run() {
 // 参加笔试
 System.out.println(Thread.currentThread().getName() + " 正在笔试...");
 sleepRandom(1000, 3000);
 System.out.println(Thread.currentThread().getName() + " 完成笔试");
 phaser.arriveAndAwaitAdvance();
 
 // 参加机试
 System.out.println(Thread.currentThread().getName() + " 正在机试...");
 sleepRandom(1500, 4000);
 System.out.println(Thread.currentThread().getName() + " 完成机试");
 phaser.arriveAndAwaitAdvance();
 
 // 参加面试
 System.out.println(Thread.currentThread().getName() + " 正在面试...");
 sleepRandom(2000, 5000);
 System.out.println(Thread.currentThread().getName() + " 完成面试");
 phaser.arriveAndDeregister(); // 退出考试
 }
 
 private void sleepRandom(int min, int max) {
 try {
 Thread.sleep(ThreadLocalRandom.current().nextInt(min, max));
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 }
 }
 }
}
```


## 高性能锁工具

### StampedLock - 乐观读锁

一种基于功能的锁，有三种模式用于控制读写访问。

StampedLock 的状态由版本和模式组成。锁获取方法返回一个戳，该戳表示并控制对锁状态的访问；这些方法的“try”版本可能会返回特殊值零，表示无法获取访问权限。锁释放和转换方法需要图章作为参数，如果它们与锁的状态不匹配，则会失败。这三种模式是：

写锁：`writeLock`方法可能会在等待独占访问时阻塞，并返回一个戳记，该戳记可在`unlockWrite`方法中使用以释放锁。还提供了无计时和计时版本的`tryWriteLock`。当锁处于写模式时，无法获取读锁，所有乐观读验证都将失败。

读锁：`readLock`方法可能会阻塞等待非独占访问，并返回一个戳记，该戳记可在`unlockRead`方法中使用以释放锁。同时，还提供了无计时和有计时的`tryReadLock`版本。

乐观读锁：方法tryOptimisticRead仅在锁当前未以写模式持有时返回非零戳记。如果自获取给定戳记以来，锁尚未以写模式被获取，则方法validate返回true，在这种情况下，最近一次写锁释放之前的所有操作都发生在调用tryOptimisticRead之后的操作之前。这种模式可以被视为一种极其弱的读锁版本，写者可以随时破坏它。对于简短的只读代码段使用乐观读取模式通常可以减少争用并提高吞吐量。然而，其使用本身具有脆弱性。乐观读取部分应仅读取字段，并将其保存在局部变量中，以便在验证后稍后使用。在乐观读取模式下读取的字段可能非常不一致，因此仅当您足够熟悉数据表示以检查一致性和/或反复调用方法validate()时，才适用此用法。例如，在首次读取对象或数组引用，然后访问其某个字段、元素或方法时，通常需要执行这些步骤。

#### 本质

乐观读锁优化

#### 原理


分为三种模式：写锁、悲观读锁、乐观读，通过戳标记（stamp）用于锁状态验证，自动进行锁升级/降级机制

#### 注意事项

- 乐观读后必须验证 stamp

- 不支持重入（与 ReentrantLock 不同）

- 避免在锁中使用条件变量

- 写锁不可直接降级为乐观读

```java
// 使用StampedLock实现高效缓存
public class OptimisticCache {
 private final StampedLock lock = new StampedLock();
 private Map<String, String> cache = new HashMap<>();
 
 // 写操作使用排他锁
 public void put(String key, String value) {
 long stamp = lock.writeLock();
 try {
 cache.put(key, value);
 } finally {
 lock.unlockWrite(stamp);
 }
 }
 
 // 读操作使用乐观锁
 public String get(String key) {
 // 1. 尝试乐观读
 long stamp = lock.tryOptimisticRead();
 String value = cache.get(key);
 
 // 2. 检查是否被修改
 if (!lock.validate(stamp)) {
 // 升级为悲观读锁
 stamp = lock.readLock();
 try {
 value = cache.get(key);
 } finally {
 lock.unlockRead(stamp);
 }
 }
 return value;
 }
 
 // 使用示例
 public static void main(String[] args) {
 OptimisticCache cache = new OptimisticCache();
 ExecutorService executor = Executors.newFixedThreadPool(10);
 
 // 并发写
 for (int i = 0; i < 5; i++) {
 final int id = i;
 executor.execute(() -> {
 cache.put("key" + id, "value" + id);
 });
 }
 
 // 并发读
 for (int i = 0; i < 10; i++) {
 executor.execute(() -> {
 for (int j = 0; j < 5; j++) {
 String value = cache.get("key" + ThreadLocalRandom.current().nextInt(5));
 System.out.println("Read: " + value);
 }
 });
 }
 
 executor.shutdown();
 }
}
```

## 总结

通过理解这些工具的核心原理和适用场景，开发者可以：

- 提升并发程序性能 3-10 倍
- 减少线程阻塞时间 60% 以上
- 降低死锁概率至 0.1% 以下
- 构建响应时间稳定的高并发系统

在调试的过程中进行调试与监控，可以使用 `JConsole` 查看锁竞争，开启 `-XX:+PrintConcurrentLocks`，或者使用 `Arthas` 监控线程阻塞

最终建议：根据实际场景组合使用工具类（如 Phaser+CompletableFuture 处理多阶段异步任务），并配合监控工具持续优化。

并发编程 - JUC 工具类的实现原理

在之前的内容中已经讲述了 JMM内存模型与volatile，同时也讲述了保障线程安全的 **同步互斥** 与 **异步非阻塞** 两种方式，更描述了异步非阻塞中的 `CAS` 原理以及 `CAS` 的一些经典问题， 本文接下来继续完成我们的并发版图，了解下什么是 `Atomic`。



## 原子类实现原理

原子类是 Java 并发编程的核心组件，位于 java.util.concurrent.atomic 包中。它们基于 CAS 机制实现无锁线程安全操作，在高并发场景下性能显著优于传统锁机制。

### 原子类核心原理 CAS

在线程安全的那篇文章中描述过什么是CAS，此处我们回顾下

#### 比较并交换

翻译：Compare-and-Swap，简称 CAS

本质就是 乐观锁思想,也是无锁的思想

#### CAS 执行流程：

CAS指令需要三个操作数，分别是变量的内存地址，用V表示，旧值用A表示，新值用B表示。

- 内存位置（V）：需要更新的变量地址
- 预期原值（A）：读取到的变量值
- 新值（B）：希望设置的新值

CAS指令在执行时，当且当V符合A时，处理器才会用B更新V的值，否则就不更新不管是否更新了V值，都会返回V的旧值，这个处理过程是一个原子操作，执行期间不会被其他线程中断

```java
// 伪代码展示 CAS 原理
boolean compareAndSwap(V expectedValue, V newValue) {
 if (this.value == expectedValue) {
 this.value = newValue;
 return true;
 }
 return false;
}
```

💡 CAS 优势：避免线程阻塞，减少上下文切换开销

⚠️ ABA 问题：值从 A→B→A 的变化无法被检测（解决方案：AtomicStampedReference）


## 原子类分类

### 基本数据类型原子类

- AtomicInteger：一个int可能原子更新的值
- AtomicLong：一个long值可以用原子更新
- AtomicBoolean：一个boolean值可以用原子更新

**常用方法**

- int addAndGet（int delta）
 
 以原子方式将输入的数值与实例中的值（AtomicInteger里的 value）相加，并返回结果

- boolean compareAndSet（int expect，int update）
 
 如果输入的数值等于预期值，则以原子方 式将该值设置为输入的值
- int getAndIncrement()

 以原子方式将当前值加1，注意，这里返回的是自增前的值
 
- int getAndSet（int newValue）

 以原子方式设置为newValue的值，并返回旧值


**核心源码分析**

```java
public class AtomicInteger {
 private volatile int value; // volatile保证可见性
 
 // Unsafe类提供CAS底层操作
 private static final Unsafe unsafe = Unsafe.getUnsafe();
 private static final long valueOffset;
 
 static {
 try {
 // 获取value字段的内存偏移地址
 valueOffset = unsafe.objectFieldOffset
 (AtomicInteger.class.getDeclaredField("value"));
 } catch (Exception ex) { throw new Error(ex); }
 }
 
 public final boolean compareAndSet(int expect, int update) {
 // 调用Unsafe的CAS指令
 return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
 }
}
```

**执行流程图**

![deepseek_mermaid_20250606_5add2a.png](https://blog.liushigong.cn/static/img/4aa9fa5cceebe7a8d5e4a10e1c028bf6.deepseek_mermaid_20250606_5add2a.webp)

**代码示例**

```java
// 原子整型计数器
AtomicInteger counter = new AtomicInteger(0);

// 原子自增（线程安全）
int newValue = counter.incrementAndGet(); // 等同于 ++i
int current = counter.getAndIncrement(); // 等同于 i++

// CAS 更新示例
boolean updated = counter.compareAndSet(5, 10); 
// 当前值=5时更新为10
```




### 数组类型原子类

- AtomicIntegerArray：一个int数组，其中元素可以原子更新
- AtomicLongArray：一个long数组，其中元素可以原子更新
- AtomicReferenceArray：可以以原子方式更新元素的对象引用数组
 
**常用方法**

- int addAndGet（int i，int delta）
 
 以原子方式将输入值与数组中索引i的元素相加。
 
- boolean compareAndSet（int i，int expect，int update）

 如果当前值等于预期值，则以原子 方式将数组位置i的元素设置成update值


**代码示例**

```java
// 原子整型数组
AtomicIntegerArray atomicArray = new AtomicIntegerArray(10);

// 原子更新数组元素
atomicArray.set(0, 100); // 设置索引0的值为100
int prev = atomicArray.getAndAdd(0, 5); // 索引0的值+5，返回旧值

// CAS 更新数组
boolean success = atomicArray.compareAndSet(0, 105, 110);
```

### 引用类型原子类

- AtomicReference：可以原子更新的对象引用
- AtomicMarkableReference：原子更新带有标记位的引用类型。可以原子更新一个布尔类 型的标记位和引用类型。
- AtomicStampedReference：：原子更新带有版本号的引用类型通过控制变量值的版本来保证CAS的正确性。解决CAS的“ABA”问题

```java
// 原子引用
AtomicReference<User> userRef = new AtomicReference<>();
User user1 = new User("Alice");
User user2 = new User("Bob");

userRef.set(user1);

// 原子更新引用
boolean updated = userRef.compareAndSet(user1, user2);

// 初始值=100, 版本号=0
AtomicStampedReference<Integer> money = new AtomicStampedReference<>(100, 0);

// 更新时检查值和版本号
int[] stampHolder = new int[1];
int currentStamp = money.getStamp();
int currentValue = money.getReference();

// 版本号+1防止ABA问题
boolean success = money.compareAndSet(
 currentValue, 
 currentValue + 50, 
 currentStamp, 
 currentStamp + 1
);
```

### 对象属性修改类型原子类

- AtomicIntegerFieldUpdater：原子更新整型的字段的更新器
- AtomicLongFieldUpdater：原子更新长整型字段的更新器
- AtomicReferenceFieldUpdate ：原子更新引用类型里的字段
 
**使用注意**

 对象的属性修改类型都是抽象类，所以在使用之前必须使用静态方法newUpdater创建一个新的更新器，然后参数传入需要更新的类和属性，属性名同时待修改的属性必须用public volatile关键字修饰
 
```java
public class AtomicIntegerFieldUpdaterTest { 
 // 创建原子更新器，并设置需要更新的对象类和对象的属性 
 private static AtomicIntegerFieldUpdater<User> a = AtomicIntegerFieldUpdater. 
 newUpdater(User.class， "old"); 
 public static void main(String[] args) { 
 // 设置柯南的年龄是10岁 
 User conan = new User("conan"， 10); 
 // 柯南长了一岁，但是仍然会输出旧的年龄 
 System.out.println(a.getAndIncrement(conan)); 
 // 输出柯南现在的年龄 
 System.out.println(a.get(conan)); 
 }
 public static class User { 
 private String name; 
 // 必须volatile修饰
 public volatile int old;
 public User(String name， int old) { this.name = name; this.old = old; }
 }
}

```


### 高性能原子类

目的是解决高并发环境下 AtomicLong 这种基于CAS算法的原子类自旋瓶颈问题，内部采用 **分段CAS机制**，高并发下性能优于 AtomicLong

- LongAccumulator：long类型的聚合器，需要传入一个long类型的二元操作，可以用来计算各种聚合操作，包括加乘等
- LongAdder：long类型的累加器，LongAccumulator的特例，只能用来计算加法，且从0开始计算
- DoubleAccumulator：double类型的聚合器，需要传入一个double类型的二元操作，可以用来计算各种聚合操作，包括加乘等
- DoubleAdder
 double类型的累加器，DoubleAccumulator的特例，只能用来计算加法，且从0开始计算
 
- Striped64：公共父类 Striped64 是实现中的核心，它实现一些核心操作，处理64位数据，很容易就能转化为其他基本类型，是个通用的类。

 
**实现思想**

采用分段的思想，提高CAS操作的成功率，AtomicLong中的成员变量value用来保存值，而在高并发的情况下value就变成了一个热点数据，也就是众多线程竞争一个 value ，将value值的操作分散到一个数组中，不同线程会命中到数组的不同槽中，各个线程只对自己槽中的那个值进行CAS操作，最后再把这些段的值相加得到最终的值。

**代码示例**

```java
// 高性能累加器（适合高并发统计）
LongAdder adder = new LongAdder();

// 多线程并发累加
parallelStream().forEach(i -> {
 adder.add(i);
});

// 获取结果（最终一致性）
long total = adder.sum();
```

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

:::tip{title="为什么会有高性能原子类"}
 AtomicLong 通过CAS算法提供了非阻塞的原子性操作，由于过多线程同时去竞争同一个变量的更新从而导致无限循环的自旋锁不停的尝试，这就是高并发下CAS性能低下的原因所在，所以推出了高性能原子类。
:::


## 适用场景

- 计数器、累加器
- 状态标志位更新
- 简单对象的状态变更
- 非阻塞算法实现


## 总结

原子类核心是基于 CAS 实现无锁线程安全，在低竞争场景性能远超传统锁，不同的场景下使用不同的类型体系：基本类型、引用类型、数组类型、对象属性修改类型，以及在高并发的场景下使用的高性能原子类类型。代码最佳实践如下：

```java
// 1. 简单计数
private AtomicInteger requestCount = new AtomicInteger(0);

// 2. 状态机更新
private AtomicReference<State> state = new AtomicReference<>(State.INIT);

// 3. 高并发统计
private LongAdder totalBytes = new LongAdder();
```

原子类作为 JUC 包的核心组件，合理使用可大幅提升并发程序性能，但需根据具体场景权衡其与锁机制的选用。

并发编程 - 深入理解 Atomic 原子类

并发编程 - 深入理解 synchronized

baidu-site-verification

本人自己的博客，对工作与学习过程中的总结、一些场景设计的实现方案、遇到过的疑难问题与解决、以及面试总结等。