这篇文章资料来自于网络是对部分知识整理这里只是记录一下仅供参考## StructuredTaskScope 介绍一、结构化并发的核心概念与设计目标Java 21 引入的结构化并发Structured Concurrency是对传统并发编程模型的重大改进。它通过明确的任务生命周期管理和作用域控制解决了长期以来困扰开发者的线程泄漏、任务状态难以追踪等问题。结构化并发的核心目标是统一并发模型将虚拟线程、平台线程、异步任务等统一到结构化作用域中。增强可观测性提供任务之间的父子关系和依赖管理。简化资源管理确保任务失败时资源的正确释放。提高代码安全性避免隐式线程泄漏和不可控的并发行为。二、结构化并发的核心组件一作用域Scopes作用域是结构化并发的核心概念用于管理一组任务的生命周期。通过Scope接口开发者可以创建子作用域通过openSubscope()方法创建嵌套作用域。启动任务使用launch()方法启动异步任务。等待任务完成通过join()方法等待所有子任务完成。处理异常通过onFailure()方法处理任务失败。import java.util.concurrent.ExecutionException; import java.util.concurrent.Scope; import java.util.concurrent.StructuredTaskScope; public class ScopeExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var task1 scope.fork(() - { System.out.println(Task 1 started); Thread.sleep(1000); System.out.println(Task 1 completed); return Result 1; }); var task2 scope.fork(() - { System.out.println(Task 2 started); Thread.sleep(2000); System.out.println(Task 2 completed); return Result 2; }); scope.join(); System.out.println(Both tasks completed); System.out.println(Task 1 result: task1.resultNow()); System.out.println(Task 2 result: task2.resultNow()); } } }二任务句柄Task Handles任务句柄代表异步执行的任务提供了以下功能获取结果resultNow()方法获取任务结果。处理异常exceptionally()方法处理任务异常。取消任务cancel()方法取消任务执行。子任务管理children()方法获取子任务句柄。import java.util.concurrent.ExecutionException; import java.util.concurrent.StructuredTaskScope; public class TaskHandleExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var parentTask scope.fork(() - { var childTask scope.fork(() - { System.out.println(Child task started); Thread.sleep(1000); System.out.println(Child task completed); return Child result; }); System.out.println(Parent task waiting for child); return childTask.resultNow(); }); scope.join(); System.out.println(Parent task result: parentTask.resultNow()); } } }三异常处理策略结构化并发提供了多种异常处理模式ShutdownOnFailure任何任务失败立即终止作用域。ContinueOnFailure允许任务继续执行收集所有异常。CustomExceptionHandler自定义异常处理逻辑。import java.util.concurrent.StructuredTaskScope; public class ExceptionHandlingExample { public static void main(String[] args) throws InterruptedException { // ShutdownOnFailure 模式 try (var scope new StructuredTaskScope.ShutdownOnFailure()) { scope.fork(() - { throw new RuntimeException(Task 1 failed); }); scope.fork(() - { System.out.println(Task 2 started); return Result 2; }); scope.join(); } catch (Exception e) { System.out.println(Caught exception: e.getMessage()); } // ContinueOnFailure 模式 try (var scope new StructuredTaskScope.ContinueOnFailure()) { scope.fork(() - { throw new RuntimeException(Task A failed); }); scope.fork(() - { throw new RuntimeException(Task B failed); }); scope.join(); System.out.println(All exceptions: scope.exceptions()); } } }三、结构化并发的高级应用技巧一任务依赖管理import java.util.concurrent.StructuredTaskScope; public class TaskDependencyExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var task1 scope.fork(() - { System.out.println(Task 1 started); Thread.sleep(1000); return Result 1; }); var task2 scope.fork(() - { System.out.println(Task 2 started); Thread.sleep(2000); return Result 2; }); var task3 scope.fork(() - { System.out.println(Task 3 started); System.out.println(Task 1 result: task1.resultNow()); System.out.println(Task 2 result: task2.resultNow()); return Result 3; }); scope.join(); System.out.println(Task 3 result: task3.resultNow()); } } }二资源管理import java.io.Closeable; import java.util.concurrent.StructuredTaskScope; public class ResourceManagementExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var resource new DatabaseConnection(); scope.fork(() - { try { resource.query(SELECT * FROM users); } finally { resource.close(); } }); scope.join(); } } static class DatabaseConnection implements Closeable { public void query(String sql) { System.out.println(Executing query: sql); } Override public void close() { System.out.println(Closing database connection); } } }三超时处理import java.time.Duration; import java.util.concurrent.StructuredTaskScope; public class TimeoutExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var task scope.fork(() - { System.out.println(Task started); Thread.sleep(3000); System.out.println(Task completed); return Result; }); scope.join(Duration.ofSeconds(2)); if (task.isDone()) { System.out.println(Task result: task.resultNow()); } else { System.out.println(Task timed out); task.cancel(); } } } }四、结构化并发的性能与内存影响一任务调度优化结构化并发通过以下方式提升性能减少线程泄漏任务自动关联作用域确保资源释放。高效的上下文切换基于虚拟线程的协作式调度。更优的内存使用避免传统线程池的固定内存开销。二与虚拟线程的协同import java.util.concurrent.Executors; import java.util.concurrent.StructuredTaskScope; public class VirtualThreadIntegrationExample { public static void main(String[] args) throws InterruptedException { try (var scope new StructuredTaskScope.ShutdownOnFailure()) { var executor Executors.newVirtualThreadPerTaskExecutor(); for (int i 0; i 10000; i) { scope.fork(() - { executor.submit(() - { System.out.println(Virtual thread task); return Result; }); return null; }); } scope.join(); } } }五、结构化并发的兼容性与迁移策略一版本兼容性结构化并发需要 JDK 21 或更高版本支持。在低版本中可以通过以下方式模拟部分功能使用 CompletableFuture手动管理任务依赖。自定义作用域类实现简单的任务生命周期管理。二迁移方案逐步重构将现有并发代码迁移到结构化作用域中。混合模式同时使用结构化并发和传统线程池。测试与监控通过单元测试和性能测试验证迁移效果。六、结构化并发的未来发展趋势一与 JVM 字节码的集成未来可能引入新的字节码指令直接支持结构化并发的生命周期管理。二框架生态的适配Spring Framework集成结构化并发的 Web 框架。Quarkus支持结构化并发的反应式扩展。Micronaut增强依赖注入与并发作用域的结合。三语言特性扩展增强的模式匹配在结构化作用域中支持更复杂的任务匹配。分布式作用域跨节点的任务生命周期管理。可视化工具支持通过 JMX 和监控工具展示结构化并发的执行情况。七、总结结构化并发是 Java 并发编程的重大突破通过明确的任务生命周期管理和作用域控制显著提升了代码的安全性和可维护性。在实际开发中结构化并发适用于以下场景需要严格资源管理的任务依赖关系复杂的并发流程分布式系统中的任务协同高并发服务中的异步处理尽管结构化并发需要 JDK 21 及以上版本支持但它已经展现出巨大的潜力。随着 Java 生态的持续优化结构化并发将成为现代 Java 开发的标准实践。合理使用结构化并发能够有效减少并发编程中的错误提高系统的可靠性和性能。ThreadForge 讲解第一步集成5分钟搞定我用的是SpringBoot项目集成特别简单就两步引依赖、加配置不用额外改代码。!-- ThreadForge核心依赖 -- dependency groupIdpub.lighting/groupId artifactIdthreadforge-core/artifactId version1.0.2/version /dependency dependency groupIdpub.lighting/groupId artifactIdthreadforge-spring-boot-starter/artifactId version1.0.2/version /dependency简单配置不用写复杂配置默认值就够用。如果想自定义超时时间、线程数在application.yml里加几行threadforge: default-timeout: 30s # 全局默认超时时间按需调整 core-pool-size: 8 # 核心线程数默认是CPU核心数不用手动算 max-pool-size: 16 # 最大线程数默认CPU核心数*2非SpringBoot项目 也能直接用不用配置直接new ThreadScope就行后面会讲。第二步核心用法3个真实场景直接抄ThreadForge的核心是ThreadScope线程作用域所有任务都要放在这个作用域里。我习惯用try-with-resources语法作用域会自动关闭不管有没有异常都能清理资源特别省心。场景1最常用——并发调用多个接口聚合结果。这是我项目里用得最多的场景比如用户详情页同时调用“查用户信息”“查订单列表”“查积分”三个接口聚合返回。用ThreadForge写就10几行代码比之前的写法少一半// 我项目里的真实业务方法直接复制能用 public UserDetailVO getUserDetail(Long userId) { // try-with-resources自动关闭作用域不用手动清理 try (ThreadScope scope ThreadScope.open()) { // 提交3个并发任务给每个任务起个名字方便排查问题 TaskUser userTask scope.submit(查询用户信息, () - userService.getById(userId)); TaskListOrder orderTask scope.submit(查询用户订单, () - orderService.listByUserId(userId)); TaskInteger pointTask scope.submit(查询用户积分, () - pointService.getByUserId(userId)); // 等待所有任务完成也可以单独等待某个任务 scope.await(userTask, orderTask, pointTask); // 获取结果组装返回 User user userTask.await(); ListOrder orders orderTask.await(); Integer point pointTask.await(); return UserDetailVO.builder() .id(user.getId()) .username(user.getUsername()) .orders(orders) .point(point) .build(); } catch (Exception e) { log.error(获取用户详情失败userId:{}, userId, e); thrownew BusinessException(获取用户详情失败请重试); } }这里有两个小细节是我踩过坑后总结的任务一定要命名submit第一个参数比如“查询用户信息”后面排查日志时能一眼看出哪个任务出问题不用对着“Task-1”猜半天不用手动处理超时默认30秒超时某个任务超时会自动取消所有任务抛出异常不用像之前那样手动写future.get(500, MILLISECONDS)。场景2进阶——批量处理部分失败不影响整体批量导入数据的场景比如导入100条用户数据部分数据导入失败不能让整个批量任务失败得汇总成功和失败的结果反馈给前端。ThreadForge有个失败策略直接用COLLECT_ALL就能收集所有任务的结果不用自己写逻辑public BatchImportResult batchImportUser(ListUserImportDTO importList) { // 打开作用域设置失败策略为COLLECT_ALL收集所有失败结果 try (ThreadScope scope ThreadScope.open() .withFailurePolicy(FailurePolicy.COLLECT_ALL) .withDeadline(Duration.ofMinutes(5))) { // 批量处理超时设5分钟 // 批量提交任务循环导入每条数据 ListTaskVoid taskList importList.stream() .map(dto - scope.submit(导入用户- dto.getUsername(), () - importUser(dto))) .collect(Collectors.toList()); // 等待所有任务完成获取结果汇总 Outcome outcome scope.await(taskList); // 组装批量结果返回给前端 return BatchImportResult.builder() .total(importList.size()) .successCount(outcome.successCount()) .failureCount(outcome.failureCount()) .failedUsernames(outcome.failedTasks().stream() .map(Task::name) .map(name - name.replace(导入用户-, )) .collect(Collectors.toList())) .build(); } } // 单个用户导入的业务方法 private void importUser(UserImportDTO dto) { // 业务逻辑校验、插入数据库 if (StringUtils.isBlank(dto.getUsername())) { thrownew IllegalArgumentException(用户名不能为空); } userService.save(User.builder() .username(dto.getUsername()) .phone(dto.getPhone()) .build()); }之前写这种批量处理得自己用CountDownLatch计数还要手动收集异常代码特别繁琐。现在用ThreadForge一行withFailurePolicy就搞定省了很多事场景3生产级用法——并发度控制监控线上调用外部接口时不能无限制并发否则会打爆对方服务同时得知道每个任务的执行情况比如耗时、成功失败数方便排查问题。ThreadForge内置了并发度控制和监控埋点不用额外集成监控工具几行代码就能实现public ListProductInfoVO batchGetProductInfo(ListLong productIds) { // 打开作用域设置最大并发度为50避免打爆外部接口 try (ThreadScope scope ThreadScope.open() .withConcurrencyLimit(50) // 任务生命周期埋点记录日志 .withHook(new ThreadHook() { Override public void onStart(TaskInfo info) { log.info(任务[{}]开始执行, info.name()); } Override public void onSuccess(TaskInfo info, Duration duration) { log.info(任务[{}]执行成功耗时{}ms, info.name(), duration.toMillis()); } Override public void onFailure(TaskInfo info, Throwable error, Duration duration) { log.error(任务[{}]执行失败耗时{}ms, info.name(), duration.toMillis(), error); } })) { // 提交任务调用外部接口获取商品信息 ListTaskProductInfoVO taskList productIds.stream() .map(id - scope.submit(获取商品信息- id, () - productApi.getProductInfo(id))) .collect(Collectors.toList()); // 等待所有任务完成获取结果 ListProductInfoVO result scope.awaitAll(taskList); // 获取监控指标打印统计信息生产环境可推送到Prometheus ScopeMetricsSnapshot metrics scope.metrics(); log.info(批量获取商品信息完成总任务{}个成功{}个失败{}个平均耗时{}ms, metrics.total(), metrics.succeeded(), metrics.failed(), metrics.averageDuration().toMillis()); return result; } catch (Exception e) { log.error(批量获取商品信息失败, e); thrownew BusinessException(获取商品信息失败); } }这个用法我在线上用了快半年从来没出过错。并发度控制能保护外部接口监控埋点能快速定位问题比如哪个商品ID调用失败耗时多久一眼就能看出来。对比一下ThreadForge到底省多少事还是以“并发调用3个接口”为例给大家看一下我之前的写法和现在的写法差距太明显了。之前的写法ExecutorServiceFuturepublic UserDetailVO getUserDetailOld(Long userId) { // 1. 手动创建线程池还要纠结核心数、队列大小 ExecutorService executor Executors.newFixedThreadPool(3); FutureUser userFuture null; FutureListOrder orderFuture null; FutureInteger pointFuture null; try { // 2. 提交任务获取Future userFuture executor.submit(() - userService.getById(userId)); orderFuture executor.submit(() - orderService.listByUserId(userId)); pointFuture executor.submit(() - pointService.getByUserId(userId)); // 3. 手动设置超时逐个获取结果 User user userFuture.get(500, TimeUnit.MILLISECONDS); ListOrder orders orderFuture.get(500, TimeUnit.MILLISECONDS); Integer point pointFuture.get(500, TimeUnit.MILLISECONDS); return UserDetailVO.builder() .id(user.getId()) .username(user.getUsername()) .orders(orders) .point(point) .build(); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error(获取用户详情失败, e); // 4. 手动取消未完成任务 if (userFuture ! null) userFuture.cancel(true); if (orderFuture ! null) orderFuture.cancel(true); if (pointFuture ! null) pointFuture.cancel(true); thrownew BusinessException(获取用户详情失败); } finally { // 5. 手动关闭线程池还要处理关闭超时 executor.shutdown(); try { if (!executor.awaitTermination(1, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); } } }60多行代码还要手动处理线程池创建、超时、取消、异常、资源清理稍微漏一步就可能出问题。我之前就因为忘了cancel任务导致线上出现无用的线程一直在跑排查了好久才找到原因。现在的写法ThreadForge就是场景1的10几行代码不用手动管理线程池不用写超时、取消逻辑不用手动关闭资源异常统一处理代码简洁可读性高就算是新手也能一次写对。我踩过的4个坑给大家避坑虽然ThreadForge很简单但我刚开始用的时候还是踩了几个坑分享给大家避免走弯路。坑1忘记用try-with-resources导致线程泄漏最开始我图省事直接new ThreadScope不关闭结果测试环境出现线程泄漏查了半天才发现是作用域没释放。错误写法// 错误不关闭scope导致资源泄漏ThreadScope scope ThreadScope.open(); TaskUser task scope.submit(查询用户信息, () - userService.getById(userId)); User user task.await();正确写法必须用try-with-resources不管是否抛出异常scope都会自动关闭坑2任务不命名排查问题困难刚开始提交任务时没给任务命名用默认的“Task-1”“Task-2”后来有个任务失败日志里只显示“Task-3执行失败”根本不知道是哪个接口出问题只能一行行排查代码。正确做法submit第一个参数一定要传入有意义的名称比如“查询用户信息-123”“获取商品信息-456”方便日志排查。坑3选错失败策略导致业务异常批量导入场景我一开始用了默认的FAIL_FAST策略一个任务失败所有任务都取消结果导入100条数据有1条失败整个批量任务都失败了被产品骂了一顿。正确做法根据业务场景选策略批量处理用COLLECT_ALL聚合接口用FAIL_FAST任务编排用SUPERVISOR。坑4不设置超时导致任务无限等待有一次做大数据量处理没设置超时时间默认30秒结果数据量太大任务超时失败。后来改成withDeadline(Duration.ofMinutes(5))就没问题了。建议每个作用域都根据业务场景设置合适的超时时间不要用默认值一刀切。最后ThreadForge不是要替代CompletableFuture、Reactor这些工具而是在日常并发场景中给我们提供一个更简单、更安全的选择。对于我们大多数Java后端开发者来说平时写的都是接口聚合、批量处理这种常规并发场景不需要极致的性能优化也不需要复杂的响应式编程ThreadForge刚好满足需求——不用费脑子少踩坑多写业务多摸鱼。我现在项目里的所有并发场景都用ThreadForge替代了传统的线程池写法代码量减少了一半排查问题也更简单上线后从来没因为多线程出过问题。如果你也经常被多线程的琐事折磨不妨试试ThreadForge相信你会跟我一样觉得“早知道这个神器就不用加那么多班了”。https://juejin.cn/post/7604779604126138368https://juejin.cn/post/7496517310787092531https://blog.csdn.net/u010581811/article/details/159397581