【八八股股 | 第三篇】CompletableFuture 与 ScopedValue：异步任务编排与上下文传递

CompletableFuture 与 ScopedValue异步任务编排与上下文传递文章摘要CompletableFuture 用来组合异步任务ScopedValue 用来在限定的调用过程中传递上下文。本文通过用户和订单查询介绍任务的提交、组合与结果获取重点对比 thenApply 和 thenCompose 返回什么、如何通过 join() 取得结果随后说明 ScopedValue 如何绑定和读取上下文、where(…).run(…) 是什么意思以及跨类使用和跨线程传递时需要注意的边界。版本说明CompletableFuture 部分以 JDK 8 API 为基础ScopedValue 部分使用 JDK 21 的预览 API需要启用--enable-preview。本文中的“预览”特指 JDK 21不用于描述所有后续 JDK 的状态。一、先区分任务编排与上下文传递假设一次请求需要完成两件事查询用户信息以及查询该用户的订单数量。如果这两个查询都能直接使用入口已有的用户 ID它们可以独立提交最后合并结果。如果必须先查询用户 ID再用 ID 查询订单后一个任务就依赖前一个任务。CompletableFuture 用来表达这些任务及其依赖关系。ScopedValue 则关注另一个问题当前用户 ID、请求标识等上下文如何在限定的调用过程中被下游代码读取而不必逐层增加方法参数。两者可以出现在同一个应用中但职责不同。使用 CompletableFuture 并不会自动获得 ScopedValue 的上下文传播能力。二、CompletableFuture表示结果也描述后续操作1. Future 能做什么FutureT表示一个计算结果可以检查是否完成、尝试取消以及通过get()获取结果。如果调用时结果还没有准备好get()会等待。[1]下面是 JDK 8 可用的代码片段假设executor已经创建FutureStringfutureexecutor.submit(()-小林);// 可以先做其他事情再在需要结果时等待Stringuserfuture.get();System.out.println(user);普通 Future 可以支持并发先提交多个任务再分别取结果即可。它的局限是Future 接口本身没有提供thenApply、thenCombine这样的后续阶段组合方法。2. CompletableFuture 增加了什么CompletableFutureT实现了FutureT和CompletionStageT。前者提供结果获取等能力后者描述计算阶段之间的依赖及后续操作。[2][3]例如下面的代码片段适用于 JDK 8CompletableFutureStringuserFutureCompletableFuture.supplyAsync(()-小林,executor);CompletableFutureStringgreetingFutureuserFuture.thenApply(user-你好user);userFuture代表用户查询结果greetingFuture代表问候语结果。调用thenApply是登记后续转换规则上一步正常完成后把它的结果交给回调。如果上一步尚未完成登记规则不需要等待它完成如果已经完成不带 Async 的回调也可能直接在登记时执行。因此登记回调不等于一定切换线程也不等于回调本身不会耗时。三、创建任务并组合没有依赖关系的查询1. runAsync 与 supplyAsyncJDK 8 中两个常用的异步创建方法分别是方法接收的任务返回类型runAsyncRunnable不返回业务值CompletableFutureVoidsupplyAsyncSupplierT返回业务值CompletableFutureT它们都可以显式传入 Executor由 Executor 安排任务执行。CompletableFuture对象是任务结果的载体本身不是一个线程。2. 用 thenCombine 合并两个查询下面是适用于 JDK 8 的完整示例。示例用固定值代替数据库查询重点展示任务之间的关系。importjava.util.concurrent.CompletableFuture;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;publicclassCombineDemo{publicstaticvoidmain(String[]args){ExecutorServiceexecutorExecutors.newFixedThreadPool(2);try{CompletableFutureStringuserFutureCompletableFuture.supplyAsync(()-小林,executor);CompletableFutureIntegerorderFutureCompletableFuture.supplyAsync(()-3,executor);CompletableFutureStringresultFutureuserFuture.thenCombine(orderFuture,(user,count)-user有count笔订单);System.out.println(resultFuture.join());}finally{executor.shutdown();}}}输出小林有3笔订单这里有三个不同动作两次supplyAsync分别提交查询任务。thenCombine登记合并规则两个阶段都正常完成后才使用两个结果执行回调。join()在输出最终结果的边界等待。thenCombine不负责启动前面两个任务。它们能否同时执行取决于线程池资源及任务调度把它们都提交到单线程池仍然只能逐个执行。3. 常用后续操作如何选择以下方法均可用于 JDK 8。表中区分了方法的用途与它本身的返回值避免把回调返回的业务值误认为整个方法的返回值。[2][3]方法用途方法本身返回什么thenApply回调使用上一步结果计算一个新值保存回调返回值的 FuturethenAccept回调使用上一步结果例如打印用户名CompletableFutureVoidthenRun执行后续操作不接收上一步结果CompletableFutureVoidthenCompose根据上一步结果继续调用异步方法代表后续任务最终结果的 FuturethenCombine两个阶段都成功后合并它们的结果保存合并结果的 FutureallOf汇合所有传入的任务不直接收集结果CompletableFutureVoid这些以then开头的方法会在相关上游阶段正常完成后执行回调上游异常时相应的回调会被跳过异常继续向后传播。thenAccept、thenRun返回的也是 Future只是结果类型是Void。“没有业务返回值”不代表“没有表示完成状态的对象”。allOf返回的 Future 会在所有传入任务完成后完成如果有任务异常完成它也会异常完成不能把“都完成”理解为“都成功”。四、thenApply 与 thenCompose返回什么怎么取结果**thenApply 和 thenCompose 本身返回的都是 CompletableFuture不会直接返回 ID 或订单。**区别在于回调返回另一个 Future 时thenApply 会多包一层thenCompose 会把后续任务接起来。[3]下面的片段适用于 JDK 8。假设userIdFuture代表用户 ID 查询queryOrdersAsync(id)返回CompletableFutureListString代表订单查询。先看三种写法写法整体返回什么最终怎么取得结果thenApply(id - id)ID 的 Futurejoin()→ IDthenApply(id - queryOrdersAsync(id))订单 Future 的 Futurejoin()→ Future再join()→ 订单thenCompose(id - queryOrdersAsync(id))订单结果的 Futurejoin()→ 订单1. thenApply回调返回 ID整体返回 ID 的 FutureCompletableFutureLongresultuserIdFuture.thenApply(id-id);LonguserIdresult.join();// 这里拿到实际 IDid - id返回的是 IDthenApply(...)返回的是CompletableFutureLong。如果回调改成id - 用户 id整体就是CompletableFutureString。回调返回什么thenApply 就把它作为新 Future 的结果。2. thenApply回调返回订单 Future整体就多包一层CompletableFutureCompletableFutureListStringresultuserIdFuture.thenApply(id-queryOrdersAsync(id));// 第一次拿到订单 FutureCompletableFutureListStringorderFutureresult.join();// 第二次拿到订单列表ListStringordersorderFuture.join();因为回调返回的是 Future所以 thenApply 的结果中有两层 Future。外层完成只表示拿到了订单 Future此时订单不一定查完。3. thenCompose接起订单查询只保留一层 FutureCompletableFutureListStringresultuserIdFuture.thenCompose(id-queryOrdersAsync(id));ListStringordersresult.join();// 一次就能拿到订单列表回调仍然返回订单 Future但 thenCompose 会跟随它的完成结果所以整体只保留一层 Future。result 仍然是 Future成功完成后通过 join() 才拿到订单列表。选择的关键是回调返回普通值还是返回另一个异步任务。不能把 thenApply 理解成普通调用、thenCompose 理解成异步调用它们本身都不保证切换线程。图 1回调返回订单 Future 时thenApply 的结果有两层 FuturethenCompose 的结果只有一层。五、线程选择、等待和异常处理1. 不带 Async不保证切换线程JDK 8 的 CompletableFuture 对不带 Async 的后续阶段不承诺统一使用某个线程池。回调可能由完成上游的线程执行上游已经完成时也可能在注册回调的线程中执行。[2]下面是 JDK 8 代码片段假设两个线程池已经创建CompletableFutureStringresultFutureCompletableFuture.supplyAsync(()-小林,ioPool).thenApply(name-你好name).thenApplyAsync(text-text,computePool);第一个任务交给ioPool中间的转换没有要求切换线程最后一个转换显式交给computePool。前一个阶段传入的线程池不会因此成为后续所有 Async 阶段自动沿用的线程池。对本文讨论的普通 CompletableFuture未显式指定 Executor 的异步方法默认使用公共池ForkJoinPool.commonPool()在 JDK 8 中公共池并行度小于 2 时会改为为每个任务创建线程。[2]耗时的阻塞 IO 任务宜使用适合业务负载的独立线程池并明确传给相应异步阶段。上文的固定线程池仅用于演示不代表生产环境的线程数与队列配置建议。2. 等待的位置会影响任务能否重叠执行下面两段均为 JDK 8 代码片段假设queryA()、queryB()返回字符串。第一种写法先等 A 完成才提交 BStringaCompletableFuture.supplyAsync(()-queryA(),executor).join();StringbCompletableFuture.supplyAsync(()-queryB(),executor).join();第二种写法先提交两个任务再等待合并结果CompletableFutureStringaCompletableFuture.supplyAsync(()-queryA(),executor);CompletableFutureStringbCompletableFuture.supplyAsync(()-queryB(),executor);Stringresulta.thenCombine(b,(x,y)-xy).join();只有第二种给两个独立任务提供了重叠执行的机会。对于本来存在依赖的任务仍然需要遵守依赖关系。get()和join()都会在必要时等待。get()需要处理InterruptedException、ExecutionException等受检异常任务异常在join()中主要以CompletionException传播。取消还可能表现为CancellationException。[1][2]3. 三种异常处理操作JDK 8 中常用的异常处理方式如下[2][3]方法触发条件用途exceptionally上游异常完成返回一个替代结果handle上游正常或异常完成根据结果或异常计算新结果whenComplete上游正常或异常完成观察结果记录日志或执行清理例如下面是一个简化的降级片段resultFuture为CompletableFutureStringCompletableFutureStringsafeFutureresultFuture.whenComplete((value,error)-{if(error!null){error.printStackTrace();// 示例日志实际项目使用日志框架}}).exceptionally(error-查询暂时不可用);whenComplete的回调不通过返回值替换业务结果通常保留上游的结果或异常但如果回调自身抛出异常也可能让后续阶段失败。实际降级还需区分可恢复异常不能把所有错误都静默转成默认值。六、ScopedValue在限定调用过程中传递上下文1. 为什么需要调用上下文一次请求中Service 可能需要用户 IDDAO 也可能需要用户 ID。如果逐层传参中间只负责转发的方法也得增加参数。ScopedValue 提供了一种隐式传递方式入口绑定上下文下游能够访问同一个键的方法在有效范围内读取它。[4]这里的“范围”是代码执行形成的动态调用范围包含直接和间接调用的方法不能只看源码中 Lambda 花括号包住了哪几行。2. 完整示例Controller → Service → DAO下面是 JDK 21 预览 API 示例。方法名称仅用于模拟分层调用不依赖 Spring 或数据库。publicclassScopedValueDemo{privatestaticfinalScopedValueLongUSER_IDScopedValue.newInstance();publicstaticvoidmain(String[]args){System.out.println(进入前已绑定USER_ID.isBound());ScopedValue.where(USER_ID,1001L).run(()-controller());System.out.println(退出后已绑定USER_ID.isBound());}staticvoidcontroller(){service();}staticvoidservice(){System.out.println(Service 用户USER_ID.get());dao();}staticvoiddao(){System.out.println(DAO 用户USER_ID.get());}}使用 JDK 21 编译和运行javac-encodingUTF-8 --enable-preview--release21ScopedValueDemo.javajava--enable-preview ScopedValueDemo输出进入前已绑定false Service 用户1001 DAO 用户1001 退出后已绑定false这段代码中三个操作分别负责不同的事情[4]newInstance()创建访问上下文的键此时还没有绑定用户 ID。where(USER_ID, 1001L).run(...)在执行传入代码期间绑定用户 ID 为 1001。USER_ID.get()读取当前绑定的用户 ID。执行结束后本次绑定自动退出即使中途抛出异常也一样。有外层绑定就恢复外层值没有则回到未绑定状态。未绑定时直接get()会抛出NoSuchElementException。图 2下游方法只要仍在有效调用范围内并且能访问同一个键就可以读取当前绑定。七、where(…).run(…) 是谁的方法必须写在同一个类吗1. 拆开链式调用下面的代码片段使用 JDK 21 预览 API与前面的链式写法等价ScopedValue.CarriercarrierScopedValue.where(USER_ID,1001L);carrier.run(()-controller());where(...)是 ScopedValue 的静态方法返回ScopedValue.Carrier。Carrier 记录键与值的映射仅创建 Carrier 尚未使绑定生效。run(...)是 Carrier 的实例方法在当前线程中执行传入的 Runnable并在这段执行期间提供绑定。[5]因此这里的run()不是 CompletableFuture 的runAsync()也不会自动创建线程。2. 绑定与读取可以位于不同类Java 不要求 ScopedValue 字段、调用where(...).run(...)的位置以及读取代码必须位于同一个类。要求是读取同一个键并处于有效范围内。下面是 JDK 21 预览 API 的跨类完整示例。为方便复制运行三个顶层类放在同一个文件中只有入口类声明为 public。publicclassContextAcrossClassesDemo{publicstaticvoidmain(String[]args){ScopedValue.where(UserContext.USER_ID,1001L).run(()-{newOrderService().createOrder();});}}finalclassUserContext{// 示例为默认包内可见其他同包类可以访问同一个键staticfinalScopedValueLongUSER_IDScopedValue.newInstance();staticlonggetUserId(){returnUSER_ID.get();}}finalclassOrderService{voidcreateOrder(){System.out.println(为用户创建订单UserContext.getUserId());}}运行方式与上节一致将文件名和入口类名替换为ContextAcrossClassesDemo。输出为为用户创建订单1001这里字段定义在 UserContext绑定发生在入口类读取发生在 OrderService它们访问的是同一个对象。如果在另一个类里重新调用ScopedValue.newInstance()即使字段也叫 USER_ID得到的仍然是另一个键不能读取前一个键的绑定。3. 用 UserContext 封装键与绑定入口实际项目可以把键设为 private只暴露需要的操作。下面是对上例的替换片段仍使用 JDK 21 预览 APIfinalclassUserContext{privatestaticfinalScopedValueLongUSER_IDScopedValue.newInstance();staticvoidrunWithUser(longuserId,Runnableaction){ScopedValue.where(USER_ID,userId).run(action);}staticlonggetUserId(){returnUSER_ID.get();}}入口改成UserContext.runWithUser(1001L,()-newOrderService().createOrder());绑定方法与字段放在同一个类是一种封装选择。下游不需要拿到键本身通过读取方法即可访问上下文。生产代码还应按职责限制哪些模块可以建立或读取用户上下文。八、ScopedValue 的常见误区与使用边界1. static 字段不代表所有线程共享同一个用户 IDstatic 字段保存的是键。绑定值按当前线程及动态范围确定。同一个键可以在不同线程的独立调用范围中分别绑定 1001、1002不会因为字段是 static 就自动相互覆盖。[4]2. 绑定不可变不等于对象自动不可变ScopedValue 没有类似ThreadLocal.set()的方法用来随意覆盖当前绑定。不过可以在嵌套动态范围内为同一个键建立新绑定退出后恢复外层值。同时绑定一个 ArrayList 并不会让列表变成只读集合。跨线程共享时仍应使用不可变对象或对可变对象的访问做适当同步。[4]3. 它与 ThreadLocal 的差别在于状态管理方式ThreadLocal 可以在线程中保存并修改状态。用于请求上下文时通常采用下面的清理方式。该片段适用于 JDK 8假设USER_ID_THREAD_LOCAL是一个ThreadLocalLong对象[6]USER_ID_THREAD_LOCAL.set(1001L);try{controller();}finally{USER_ID_THREAD_LOCAL.remove();}线程池会复用线程。如果上下文未按请求边界清理后续任务可能读到残留值。ScopedValue 则把绑定有效期与一次受控调用关联起来退出时自动恢复。对于从调用方单向传给下游的上下文ScopedValue 提供了更明确的生命周期这不意味着所有需要可变线程状态的 ThreadLocal 场景都能原样替换。4. 普通线程池任务不会自动继承绑定前面的 Controller → Service → DAO 示例全部在同一个线程执行。它已经能够体现 ScopedValue 的用途并不要求切换线程。下面是 JDK 21 预览 API 的误用片段假设 USER_ID 只在提交任务的线程中绑定executor 是普通线程池ScopedValue.where(USER_ID,1001L).run(()-{CompletableFutureLongfutureCompletableFuture.supplyAsync(()-USER_ID.get(),executor);future.join();});不能因为提交任务时存在绑定就认为工作线程也能读取它。普通线程池不会自动复制该绑定工作线程中未绑定时get()会失败。是否传播与提交发生在花括号内部没有必然关系。JDK 21 支持通过StructuredTaskScope进行结构化继承创建任务范围时捕获绑定由该范围通过fork()启动的子任务继承父子任务的生命周期受到约束。StructuredTaskScope在 JDK 21 同样是预览 API。[4][7]这说明跨线程传递上下文需要明确的传播机制。CompletableFuture 管理任务结果和依赖ScopedValue 管理有界上下文不能把其中一个的能力自动归给另一个。推荐标签Java、CompletableFuture、异步编程、ScopedValue、Java 21参考资料Java SE 8Future API——结果获取、等待及异常约定。Java SE 8CompletableFuture API——异步创建、默认执行策略、汇合及结果获取。Java SE 8CompletionStage API——后续阶段、组合规则及异常传播。Java SE 21ScopedValue API——动态作用域、重绑定、线程隔离及继承。Java SE 21ScopedValue.Carrier API——映射与 run/call 的执行语义。Java SE 21ThreadLocal API——线程本地状态及 set/remove 操作。Java SE 21StructuredTaskScope API——结构化子任务与 ScopedValue 继承。Java SE 21 Core LibrariesScoped Values——JDK 21 预览功能说明。