当前位置：首页 > ds >正文

Java 并发新范式：用 Structured Concurrency 优雅收拾多线程烂摊子

ds 2025/8/16 18:41:12

1. 背景：为什么要有 Structured Concurrency

传统多线程（Thread + ExecutorService）的问题：

线程生命周期难管理：启动的线程分散在不同地方，不容易跟踪，也不容易知道什么时候全部完成。
错误传播复杂：一个线程抛异常时，调用方往往无感知，必须自己加监听或 future.get()。
资源泄漏风险：启动的任务可能在调用方结束后还在跑，导致内存泄漏或数据错乱。
可读性差：任务之间的逻辑关系不直观。

Structured Concurrency 借鉴了 Go、Kotlin（coroutineScope）等语言的理念，把并发任务看作一个整体结构来管理，保证：

所有子任务的生命周期受父任务控制。
任务异常能自动传播给父任务。
任务必须在作用域结束前全部完成（或取消）。

2. 概念

Structured Concurrency 核心思想：

“并发任务的生命周期，应该和它们的作用域一致。”

Java 在 JEP 453（JDK 21 预览）和 JEP 462（JDK 22 第二次预览）中实现了该特性。

类位置：

java.util.concurrent.StructuredTaskScope

常用子类：

StructuredTaskScope.ShutdownOnFailure
任一任务失败，就取消其它任务，父任务抛异常。
StructuredTaskScope.ShutdownOnSuccess
任一任务成功，就取消其它任务，返回该结果。

3. 基本用法示例

假设我们需要并发调用两个远程接口，取最快返回的结果：

import java.util.concurrent.*;public class StructuredConcurrencyExample {public static void main(String[] args) throws InterruptedException, ExecutionException {try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {scope.fork(() -> fetchFromServerA()); // 子任务1scope.fork(() -> fetchFromServerB()); // 子任务2scope.join();   // 等待所有任务结束（成功或被取消）String result = scope.result(); // 获取第一个成功的结果System.out.println("结果: " + result);}}static String fetchFromServerA() throws InterruptedException {Thread.sleep(2000);return "A的数据";}static String fetchFromServerB() throws InterruptedException {Thread.sleep(1000);return "B的数据";}
}

输出：

结果: B的数据

特点：

try-with-resources 确保作用域结束时自动清理。
不用手动管理 Future 或线程取消。
异常会自动向外传播。

4. 错误传播与取消

如果用 ShutdownOnFailure，任一任务失败就会：

取消其它任务
把异常抛给父任务

示例：

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {scope.fork(() -> {Thread.sleep(1000);return "OK";});scope.fork(() -> {throw new RuntimeException("任务失败");});scope.join();       // 会在这里感知异常scope.throwIfFailed(); // 把异常向上传递
}

5. 对比传统 ExecutorService

传统写法：

var executor = Executors.newFixedThreadPool(2);
Future<String> f1 = executor.submit(this::task1);
Future<String> f2 = executor.submit(this::task2);
String r1 = f1.get();
String r2 = f2.get();
executor.shutdown();