Java 21结构化并发实战：再也不用手动管理线程池了

上个星期我用Java 21重构了一个老项目里的并发逻辑。原来是用CompletableFuture + 手动线程池去同时调用三个第三方接口，再汇总结果。写的时候头大，试了很多姿势，finally里加future.cancel()，还要处理各种异常，代码绕来绕去。后来看到JDK 21正式加入了StructuredTaskScope，也就是结构化并发。花了一下午把那段逻辑重写了一遍，代码量直接少了三分之一，关键是出问题的时候好排查了。

这篇文章拿一个实际场景来演示：APP首页需要同时获取用户信息、今日推荐列表、系统通知，三个接口可能只是一个个RPC调用，但处理逻辑类似。我会先写上古老的写法，再写结构化并发的写法，顺便说说踩坑记录。

场景设定

假设我们有一个HomeService，需要给客户端返回一个聚合对象，里面包含以下内容：

• 用户基本信息（UserInfo）
• 根据用户兴趣推荐的商品列表（List<Product>）
• 系统公告（Notice）

三个数据源互相独立，所以应该并行去取。以前用ExecutorService的话，大概会这么写：

ExecutorService executor = Executors.newFixedThreadPool(3);

Future<UserInfo> userFuture = executor.submit(() -> fetchUserInfo());
Future<List<Product>> productFuture = executor.submit(() -> fetchProducts());
Future<Notice> noticeFuture = executor.submit(() -> fetchNotice());

try {
    UserInfo user = userFuture.get(2, TimeUnit.SECONDS);
    List<Product> products = productFuture.get(2, TimeUnit.SECONDS);
    Notice notice = noticeFuture.get(2, TimeUnit.SECONDS);
    return new HomeData(user, products, notice);
} catch (Exception e) {
    // 其中一个超时或失败，要么全部取消，要么记录日志
    userFuture.cancel(true);
    productFuture.cancel(true);
    noticeFuture.cancel(true);
    throw new RuntimeException("聚合查询失败", e);
} finally {
    executor.shutdown();
}

这个写法最大的问题是：如果三个任务都成功，那没问题。但如果第一个future超时了，你会取消另外两个，这个逻辑没问题。可是如果第一个future抛异常，第二个future还在正常运行，你虽然取消了，但可能已经造成了线程资源浪费。还有一个更麻烦的事：线程池是手动创建的，如果忘记shutdown，整个应用可能直接崩掉。

再一个痛点是错误信息不直观。哪一步失败，怎么失败的，你都得靠断点和日志去猜。特别是并发任务多的时候，查找问题就像大海捞针。

结构化并发：把并发任务装进一个“代码块”

结构化并发的核心思想是：任务的生命周期与代码块一致。你创建一个StructuredTaskScope，在作用域里提交子任务，最后在作用域结束的地方统一join。如果其中某个子任务失败，scope会取消其他未完成的任务。就像写顺序代码一样，结构清晰，自动管理生命周期。

用StructuredTaskScope重写上面的逻辑，代码变成这样：

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {

    Subtask<UserInfo> userSubtask = scope.fork(() -> fetchUserInfo());
    Subtask<List<Product>> productSubtask = scope.fork(() -> fetchProducts());
    Subtask<Notice> noticeSubtask = scope.fork(() -> fetchNotice());

    scope.join();
    scope.throwIfFailed(e -> new RuntimeException("聚合查询失败", e));

    UserInfo user = userSubtask.get();
    List<Product> products = productSubtask.get();
    Notice notice = noticeSubtask.get();

    return new HomeData(user, products, notice);
}

ShutdownOnFailure 这个策略的意思是：只要有一个子任务失败，就取消其他未完成的子任务，然后抛出异常。你也可以用 ShutdownOnSuccess，意思是只要一个成功就返回，适合“谁先返回用谁”的场景。

你自己看，这个写法是不是比Future清爽很多？最关键的是不需要再手动管理线程池了。因为StructuredTaskScope内部会创建一个虚拟线程池，并且保证所有子任务在scope关闭前都结束。用完scope自动关闭，不会留下游离线程。

更细一点：Subtask的获取细节

fork返回的Subtask对象，你可以理解成一个“有结果的句柄”。在scope.join()之前不能调用get()，否则会报错。只有join之后才安全。

而且join()不是无脑等待，它有两个变体：join()等待所有任务完成（不管是成功还是失败），join(Duration)只等待一段时间，超时后如果还有任务没结束，那么scope的状态就会变成“未完成”。这时候如果你用ShutdownOnFailure，它会自动cancel掉还在跑的任务。

有一次我没加超时，然后其中一个RPC服务卡死了，整个接口也卡住不动，直到上游timeout才返回。后来我改成这样：

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    // fork...
    scope.join(Duration.ofSeconds(2));
    scope.throwIfFailed(e -> new RuntimeException("聚合查询失败", e));

    if (!userSubtask.isCompleted() || !productSubtask.isCompleted() || !noticeSubtask.isCompleted()) {
        throw new RuntimeException("查询超时");
    }
}

这样就能控制整体的超时时间，而不用分别给每个future设置超时。而且写法比多个future individually try-catch舒服多了。

搭配虚拟线程才是最佳拍档

StructuredTaskScope底层默认使用虚拟线程执行fork的任务。虚拟线程的创建成本极低，所以即使同时fork几百个任务也没多大压力。在我这个场景里只有三个任务，感受不明显。但如果你要循环遍历一个列表，为每一个item并发去调用远程接口，那虚拟线程的价值就体现出来了。

举个例子，以前用CompletableFuture + fixedThreadPool(16)要小心队列积压，现在直接：

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    List<Subtask<ItemData>> subtasks = itemList.stream()
            .map(item -> scope.fork(() -> fetchItemDetail(item)))
            .toList();

    scope.join();
    scope.throwIfFailed(ex -> new RuntimeException("拉取明细失败", ex));

    return subtasks.stream()
            .map(Subtask::get)
            .toList();
}

这代码是不是很惊艳？没有线程池、没有countDownLatch、没有future list。而且由于虚拟线程很轻，你甚至不需要限制并发数量，完全可以把整个列表全部丢进去。不过还是要小心下游服务的承受能力。

异常处理的一个坑：运行时异常会被包装

我的子任务里如果抛的业务异常是自定义的CheckException，fork的时候不会立即抛出来，而是会用scope.throwIfFailed转一圈。throwIfFailed里接收一个Function，参数是异常对象，你可以在这里重新包装或者直接返回原异常。但如果你直接不调用throwIfFailed，异常就会被吞掉，Subtask.get()也不会抛出，而是返回null或者卡住。所以记住：join之后必须调用throwIfFailed。

另外，如果子任务里抛出的异常本身就是Error（比如OutOfMemoryError），那是别的情况。反正普通异常没问题。

结构化并发不适用什么场景

它不适合那种“发给后台就不管了”的异步任务。比如你提交一个日志批量写入任务，不期望等它结果。这种场景用StructuredTaskScope就不合适，因为scope退出时会等待所有任务完成。所以它适合需要聚合多个结果、且能清晰定义任务生命周期的场景。

还有就是不要在fork的任务里再去创建新的StructuredTaskScope，这会导致结构嵌套过度，而且外层scope不会等待内层scope的子任务。最好保持平铺结构。

最终的HomeService长这样

public HomeData getHomeData(Long userId) {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {

        Subtask<UserInfo> userSubtask = scope.fork(() -> userClient.fetchInfo(userId));
        Subtask<List<Product>> productSubtask = scope.fork(() -> recommendClient.fetchProducts(userId));
        Subtask<Notice> noticeSubtask = scope.fork(() -> noticeClient.fetchLatest());

        scope.join(Duration.ofSeconds(2));
        scope.throwIfFailed(e -> new RuntimeException("首页聚合接口异常", e));

        if (!userSubtask.isCompleted() || !productSubtask.isCompleted() || !noticeSubtask.isCompleted()) {
            throw new RuntimeException("首页聚合接口超时");
        }

        return new HomeData(
                userSubtask.get(),
                productSubtask.get(),
                noticeSubtask.get()
        );
    }
}

代码就这么干净。没有一长串future.cancel()，也没有线程池优雅关闭的问题。最让我喜欢的一点是，如果你在调试时看到某个任务一直没返回，在IDE里展开scope，能清楚看到还有几个子任务没结束，都是什么状态，比之前查Future好看多了。

说几句大实话

结构化并发并不是什么黑魔法，它就是把并发任务的作用域变得更明确，让代码读起来像顺序执行一样。如果你手下的项目还是JDK17，那暂时用不了，得升到JDK21。可能会担心升级成本，但我自己用下来，真正影响代码兼容的其实很少。新项目我建议直接上JDK21，老旧项目还是老老实实用CompletableFuture吧，别乱折腾。

最后，如果你也打算在项目里用StructuredTaskScope，建议把超时和异常处理提前设计好。这玩意儿比Future好用，但也好用到容易让人忘记边界。