结构化并发要在 JDK 28 转正了。JEP 543 在 10 月 6 日进入 Proposed to Target，明确写着「不再做任何改动」直接定稿。我拿 JDK 28 EA build 19 配 Spring Boot 4.1.1 写了个聚合接口，失败取消、超时、竞速、线程转储都实际跑了一遍，行为和 JEP 描述一致，只是这个 EA 版本里还得加 --enable-preview。
主源：JEP 543: Structured Concurrency，状态变化也可以在 JDK 28 项目页 跟踪。
走了多久
这个 API 从 JDK 19 开始孵化，到 JDK 28 一共经历了两轮孵化、七轮预览。中间改过几次大的：JDK 21 把 fork 的返回值从 Future 换成 Subtask，JDK 25 把公开构造器换成静态 open(...) 工厂方法，JDK 27 给 join() 加了异常类型参数。最终定下来的接口是这样：
public sealed interface StructuredTaskScope<T, R, R_X extends Throwable>
extends AutoCloseable {
static <T> StructuredTaskScope<T, Void, ExecutionException> open();
static <T, R, R_X extends Throwable> StructuredTaskScope<T, R, R_X> open(
Joiner<? super T, ? extends R, R_X> joiner);
<U extends T> Subtask<U> fork(Callable<? extends U> task);
Subtask<? extends T> fork(Runnable task);
R join() throws R_X, InterruptedException;
void close();
}
T 是子任务结果类型，R 是 join() 的返回类型，R_X 是 join() 抛的异常类型。JDK 27 的面试题里也聊过它的预览形态，见 Java 27 面试怎么答。
环境
- JDK：
openjdk version "28-ea" 2027-03-23，build 28-ea+19-1485，来自 jdk.java.net/28 - Spring Boot 4.1.1（spring-boot-starter-webmvc，内嵌 Tomcat 11.0.24）
- Maven 3.9.9，Linux x64
先确认 build 19 里它还是预览 API。不加参数直接跑会报错：
T.java:1: error: StructuredTaskScope is a preview API and is disabled by default.
import java.util.concurrent.StructuredTaskScope;
^
(use --enable-preview to enable preview APIs)
1 error
JEP 状态现在是 Proposed to Target，还没有 Integrated，所以 build 19 里它还带着预览标记。
项目配置
pom.xml 里把 java.version 设成 28，编译器加上 --enable-preview：
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>4.1.1</version>
<relativePath/>
</parent>
<properties>
<java.version>28</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webmvc</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<compilerArgs>
<arg>--enable-preview</arg>
</compilerArgs>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
application.properties 打开虚拟线程，日志格式里带上线程名，方便看子任务跑在哪：
spring.threads.virtual.enabled=true
logging.pattern.console=%d{HH:mm:ss.SSS} %-5level [%15.15thread] %-28.28logger{0} : %msg%n
模拟下游服务
用 Thread.sleep 代替网络调用，被中断时打一行 cancelled，这样日志里能直接看到谁被取消了：
@Component
public class Downstream {
private static final Logger log = LoggerFactory.getLogger(Downstream.class);
public String user(String id, long ms, boolean fail) throws InterruptedException {
return call("user", ms, fail, "user-" + id);
}
public int orderCount(String id, long ms, boolean fail) throws InterruptedException {
return call("order", ms, fail, 3);
}
public int points(String id, long ms, boolean fail) throws InterruptedException {
return call("points", ms, fail, 1280);
}
public String quote(String region, long ms) throws InterruptedException {
return call("quote-" + region, ms, false, region + ":42.0");
}
private <T> T call(String name, long ms, boolean fail, T value) throws InterruptedException {
log.info("{} start, rid={}", name, RequestContext.RID.orElse("-"));
try {
Thread.sleep(Duration.ofMillis(ms));
} catch (InterruptedException e) {
log.info("{} cancelled", name);
throw e;
}
if (fail) {
log.info("{} failed", name);
throw new IllegalStateException(name + " service unavailable");
}
log.info("{} done in {}ms", name, ms);
return value;
}
}
RequestContext.RID 是个 ScopedValue，用来验证子任务能不能拿到父任务绑定的请求 ID：
public final class RequestContext {
public static final ScopedValue<String> RID = ScopedValue.newInstance();
private RequestContext() {}
}
聚合接口
/dashboard/{id} 并发查用户、订单数、积分，三个调用分别耗时 200、300、800 毫秒。fail 参数让某个服务抛异常，timeoutMs 给整个 scope 设超时：
@GetMapping("/dashboard/{id}")
public ResponseEntity<?> dashboard(@PathVariable String id,
@RequestParam(defaultValue = "") String fail,
@RequestParam(defaultValue = "0") long timeoutMs) throws Exception {
String rid = UUID.randomUUID().toString().substring(0, 8);
return ScopedValue.where(RequestContext.RID, rid).call(() -> load(id, fail, timeoutMs));
}
private ResponseEntity<?> load(String id, String fail, long timeoutMs) throws InterruptedException {
long start = System.nanoTime();
try (var scope = StructuredTaskScope.open(cf -> {
var c = cf.withName("dashboard-" + id);
return timeoutMs > 0 ? c.withTimeout(Duration.ofMillis(timeoutMs)) : c;
})) {
Subtask<String> user = scope.fork(() -> ds.user(id, 200, fail.equals("user")));
Subtask<Integer> orders = scope.fork(() -> ds.orderCount(id, 300, fail.equals("order")));
Subtask<Integer> points = scope.fork(() -> ds.points(id, 800, fail.equals("points")));
scope.join();
long ms = (System.nanoTime() - start) / 1_000_000;
return ResponseEntity.ok(new Dashboard(user.get(), orders.get(), points.get(), ms));
} catch (ExecutionException e) {
long ms = (System.nanoTime() - start) / 1_000_000;
return ResponseEntity.status(503).body(Map.of(
"error", e.getCause().getClass().getSimpleName(),
"message", String.valueOf(e.getCause().getMessage()),
"elapsedMs", ms));
}
}
这里用的是 open(UnaryOperator)，默认策略是任何一个子任务失败就取消其他子任务。withName 给 scope 起名，后面在线程转储里能看到。
正常请求
$ curl -s localhost:8080/dashboard/42
{"user":"user-42","orders":3,"points":1280,"elapsedMs":804}
总耗时 804 毫秒，基本等于最慢的那个调用。日志里三个子任务同一毫秒启动，各自跑在虚拟线程上，rid 都继承到了：
11:31:26.370 INFO [ virtual-54] Downstream : points start, rid=7df59210
11:31:26.370 INFO [ virtual-52] Downstream : order start, rid=7df59210
11:31:26.370 INFO [ virtual-51] Downstream : user start, rid=7df59210
11:31:26.571 INFO [ virtual-51] Downstream : user done in 200ms
11:31:26.671 INFO [ virtual-52] Downstream : order done in 300ms
11:31:27.171 INFO [ virtual-54] Downstream : points done in 800ms
一个失败，其余取消
订单服务在 300 毫秒时抛异常：
$ curl -s "localhost:8080/dashboard/42?fail=order"
{"message":"order service unavailable","error":"IllegalStateException","elapsedMs":302} HTTP 503
302 毫秒就返回了，没有等积分服务的 800 毫秒。日志能看出来，order 失败后 1 毫秒，还在跑的 points 就被中断了：
11:31:27.234 INFO [ virtual-60] Downstream : order start, rid=e95bac84
11:31:27.234 INFO [ virtual-61] Downstream : points start, rid=e95bac84
11:31:27.234 INFO [ virtual-59] Downstream : user start, rid=e95bac84
11:31:27.434 INFO [ virtual-59] Downstream : user done in 200ms
11:31:27.534 INFO [ virtual-60] Downstream : order failed
11:31:27.535 INFO [ virtual-61] Downstream : points cancelled
换成 ExecutorService 加 Future.get() 的老写法，points 会在后台继续跑完，这就是 JEP 里说的线程泄漏。
超时
给 scope 设 500 毫秒超时：
$ curl -s "localhost:8080/dashboard/42?timeoutMs=500"
{"message":"null","error":"CancelledByTimeoutException","elapsedMs":502} HTTP 503
join() 抛出 ExecutionException，cause 是 CancelledByTimeoutException，和 JEP 描述一致。没跑完的 points 同样被取消：
11:31:27.558 INFO [ virtual-65] Downstream : user start, rid=90ded819
11:31:27.559 INFO [ virtual-67] Downstream : points start, rid=90ded819
11:31:27.559 INFO [ virtual-66] Downstream : order start, rid=90ded819
11:31:27.759 INFO [ virtual-65] Downstream : user done in 200ms
11:31:27.860 INFO [ virtual-66] Downstream : order done in 300ms
11:31:28.060 INFO [ virtual-67] Downstream : points cancelled
用 Joiner 换策略
三个冗余报价源，谁先回来用谁，用 Joiner.anySuccessfulOrThrow()：
@GetMapping("/quote")
public Map<String, Object> quote() throws Exception {
long start = System.nanoTime();
try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow())) {
scope.fork(() -> ds.quote("sh", 400));
scope.fork(() -> ds.quote("hk", 150));
scope.fork(() -> ds.quote("sg", 600));
String winner = scope.join();
return Map.of("quote", winner, "elapsedMs", (System.nanoTime() - start) / 1_000_000);
}
}
$ curl -s localhost:8080/quote
{"quote":"hk:42.0","elapsedMs":152}
hk 在 150 毫秒拿到结果，另外两个立刻被取消：
11:31:28.073 INFO [ virtual-71] Downstream : quote-hk start, rid=-
11:31:28.073 INFO [ virtual-70] Downstream : quote-sh start, rid=-
11:31:28.073 INFO [ virtual-72] Downstream : quote-sg start, rid=-
11:31:28.223 INFO [ virtual-71] Downstream : quote-hk done in 150ms
11:31:28.224 INFO [ virtual-72] Downstream : quote-sg cancelled
11:31:28.224 INFO [ virtual-70] Downstream : quote-sh cancelled
结果类型都一样、想直接拿列表的时候，用 allSuccessfulOrThrow()，join() 直接返回 List，不用再挨个 Subtask.get()：
@GetMapping("/quotes")
public Map<String, Object> quotes() throws Exception {
try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow())) {
for (var r : List.of("sh", "hk", "sg")) {
scope.fork(() -> ds.quote(r, 100));
}
return Map.of("quotes", scope.join());
}
}
$ curl -s localhost:8080/quotes
{"quotes":["sh:42.0","hk:42.0","sg:42.0"]}
线程转储里能看到 scope
JEP 543 把 JSON 格式的线程转储扩展了，scope 会作为线程容器出现。我在一次 dashboard 请求还没结束时执行：
jcmd <pid> Thread.dump_to_file -format=json td.json
从 td.json 里摘出名字以 dashboard 开头的容器（只截了栈顶 6 帧）：
{
"container": "dashboard-7/jdk.internal.misc.ThreadFlock$ThreadContainerImpl@28b967c9",
"parent": "<root>",
"owner": 48,
"threadCount": 1,
"threads": [
{
"tid": 54,
"name": "",
"state": "TIMED_WAITING",
"stack": [
"java.base/java.lang.VirtualThread.parkNanos(VirtualThread.java:807)",
"java.base/java.lang.VirtualThread.sleepNanos(VirtualThread.java:994)",
"java.base/java.lang.Thread.sleepNanos(Thread.java:585)",
"java.base/java.lang.Thread.sleep(Thread.java:679)",
"com.example.sc.Downstream.call(Downstream.java:34)",
"com.example.sc.Downstream.points(Downstream.java:24)"
]
}
]
}
withName 设的 dashboard-7 直接出现在容器名里，owner 是处理请求的那个线程。这时 user 和 order 已经结束，容器里只剩还在睡的 points，一个请求卡在哪个下游，从容器名和栈顶就能看出来。
JEP 里写明的几条约束
取消靠的是中断，所以子任务要响应中断，阻塞在不可中断调用上的子任务会让 close() 一直等。fork 只能由打开 scope 的线程调用，join 也要由它在 scope 内调用；scope 不放在 try-with-resources 里、或者嵌套关闭的顺序不对，会抛 StructureViolationException。
Joiner 每个 scope 都要新建，不能复用，Subtask.get() 只能在 join() 之后调用。StructuredTaskScope 也没有实现 ExecutorService，原来往方法里传 executor 的代码，要改成在方法内部开 scope。
参考：JEP 543: Structured Concurrency、JDK 28 Early-Access Builds、JDK 28 项目页。
一起交流
分享你的思考，让讨论更进一步。