这次我们不看模型部署也不聊显卡显存而是回到 Java 后端开发里一个非常典型的工程问题顺序工作流异步执行。简单说就是有一串任务必须按顺序执行但每个任务都可能是耗时的 IO 操作、远程调用或者计算密集操作。如果你把它们串在一个线程里同步跑整体耗时会累加系统吞吐也上不去。如果拆成多个线程并行跑又无法保证“前一个任务结束后才能启动后一个任务”的依赖关系。更麻烦的是当中间某个任务抛出异常时后续任务到底该不该继续执行很多团队在这里踩过坑——一个任务异常后整个工作流却还在继续跑结果数据错乱或者重复触发下游操作。这篇文章会围绕CompletableFuture展开重点解决三个问题顺序工作流的异步编排怎么保证任务按序执行又不浪费线程资源。某个异步任务异常后如何让后续任务“不再执行”并让异常在调用链中正确传播。工作流批量执行、接口 API 封装、性能观察和排查方法。先给一个核心能力速览方便你判断这篇文章适不适合你。1. 核心能力速览能力项说明所属领域Java 后端并发编程 / 异步任务编排核心 APICompletableFutureJDK 8主要解决顺序工作流的异步化、异常传播、任务编排核心场景订单状态流转、审批流、数据同步管道、多步骤任务处理是否支持批量任务支持可结合线程池和任务队列批量执行是否支持接口 API支持可封装为 REST API 或消息队列消费者是否需要 GPU不需要推荐 JDKJDK 17 或 JDK 21长期支持版本推荐构建工具Maven 或 Gradle从技术栈看这个主题不挑机器、不挑环境只要 JDK 8 就能跑。但如果你要用到虚拟线程Thread.ofVirtual()那需要 JDK 21。建议直接上 JDK 17性能和稳定性比较平衡。2. 顺序工作流异步执行到底解决什么问题先明确一个概念。所谓“顺序工作流”指的是任务之间存在依赖关系必须按序执行。比如任务 A校验参数 - 任务 B查询用户信息 - 任务 C生成预览数据 - 任务 D写入数据库这种流程在业务系统里很常见。但问题在于每个任务消耗的时间不同。如果同步执行整个链路耗时是所有任务耗时之和。如果任务 D 依赖于任务 C 的返回值任务 C 又依赖于任务 B 的返回值那么天然无法并行只能顺序执行。这时候有两条路每个任务一个线程池task A 结束之后手动提交 task B依靠线程切换和回调机制实现异步。用CompletableFuture把任务串联起来让每个任务在独立的线程中执行任务完成后自动触发下一个任务。方案 1 的问题在于回调嵌套。业务环节一多代码就变成“回调地狱”而且很难处理异常。方案 2 是CompletableFuture的典型应用场景用一个CompletableFuture表示一个异步任务的结果通过thenApply、thenCompose、handle等方法把多个异步任务串成链。这里需要重点区分两个 APIthenApply和thenCompose。方法返回值适用场景thenApplyCompletableFutureU但函数返回的是直接结果 U同步转换任务结果thenComposeCompletableFutureU且函数返回的是CompletableFutureU异步任务依赖另一个异步任务避免嵌套如果任务 B 本身就是异步操作比如远程调用返回CompletableFutureOrder那么应该用thenCompose否则会得到CompletableFutureCompletableFutureOrder这种嵌套结构。这是很多初学者容易踩的坑。顺序工作流异步执行的核心思路就是利用CompletableFuture的链路传播机制让每个阶段都在线程池中运行阶段之间通过回调自动衔接。3. CompletableFuture 基础几个必须掌握的原子能力在开始编排顺序工作流之前先确认你对以下 API 的掌握程度。这些都是后面代码示例的基础。3.1 创建异步任务CompletableFutureString future CompletableFuture.supplyAsync(() - { // 这里是异步任务 return result; }, executorService);supplyAsync适合有返回值的任务。如果不需要返回值可以使用runAsync。3.2 串行依赖CompletableFutureString future CompletableFuture .supplyAsync(() - result1, executorService) .thenApplyAsync(result1 - result1 -result2, executorService) .thenApplyAsync(result2 - result2 -result3, executorService);thenApplyAsync表示转换操作也在线程池中异步执行。如果省略Async任务会在上一个任务执行的线程中同步执行不一定能提升吞吐。3.3 异常处理CompletableFuture提供三类异常处理方法方法行为exceptionally只处理异常返回一个默认值链会继续handle无论成功失败都会执行返回新的结果whenComplete消费结果或异常不能改变最终结果这里的关键问题是异常之后后续任务还会不会执行答案是默认情况下CompletableFuture链中某个任务抛出异常后续依赖它的任务会直接跳过并收到异常信息。也就是说如果你使用thenApply串联任务前面的任务抛异常后后面的thenApply不会执行。但是有个坑如果你在中间使用了exceptionally或者handle并且返回了一个默认值那么异常就被“吞掉”了后续任务会继续执行。这既是能力也是风险。4. 顺序工作流的三种异步编排方式根据任务之间的依赖关系和异常处理策略顺序工作流可以分成三种编排方式。下面逐个说明并附上代码示例。4.1 线性链路thenApply / thenCompose 串联这种方式适合任务之间严格串行、且后一个任务依赖前一个任务返回值的场景。public CompletableFutureWorkflowResult executeLinearWorkflow() { return CompletableFuture .supplyAsync(() - step1(), executorService) .thenApplyAsync(result1 - step2(result1), executorService) .thenApplyAsync(result2 - step3(result2), executorService) .thenApplyAsync(result3 - new WorkflowResult(result3), executorService); }这种写法最直观。测试时如果step2抛出异常step3不会执行最终返回的CompletableFuture会以CompletableFuture.completedExceptionally结束。调用方可以通过get()或join()收到异常。4.2 依赖分支thenCompose 处理嵌套异步任务如果step2内部的业务逻辑也是异步的比如调用了远程服务返回CompletableFutureResult2此时应该使用thenComposepublic CompletableFutureWorkflowResult executeNestedWorkflow() { return CompletableFuture .supplyAsync(() - step1(), executorService) .thenComposeAsync(result1 - asyncStep2(result1), executorService) .thenComposeAsync(result2 - asyncStep3(result2), executorService) .thenApplyAsync(result3 - new WorkflowResult(result3), executorService); }thenComposeAsync不仅消除了嵌套的CompletableFuture还能保证链路执行顺序。异常处理逻辑和线性链路一致中途异常后续任务跳过。4.3 并行 汇聚同时执行多个前置任务顺序工作流并不是说整条链路只有一个线程在跑。多个互不依赖的前置任务可以先并行执行然后再进入下一个顺序阶段。public CompletableFutureWorkflowResult executeParallelThenSequential() { CompletableFutureResultA futureA CompletableFuture.supplyAsync(() - taskA(), executorService); CompletableFutureResultB futureB CompletableFuture.supplyAsync(() - taskB(), executorService); return futureA .thenCombineAsync(futureB, (resultA, resultB) - { // 两个前置任务都完成后在这里合并结果再进入下一阶段 return mergeAndNext(resultA, resultB); }, executorService) .thenComposeAsync(merged - asyncStep(merged), executorService); }thenCombine会在两个CompletableFuture都完成之后执行合并逻辑。如果其中一个异常合并阶段会直接收到异常后续的asyncStep不会执行。这张表格总结了三种编排方式的适用场景编排方式核心 API适用场景异常后行为线性链路thenApplyAsync严格串行任务间有数据传递后续任务跳过异常向上传播依赖分支thenComposeAsync某个任务本身是异步操作后续任务跳过异常向上传播并行 汇聚thenCombineAsync多个前置任务互相独立等待所有任务异常则跳过后续5. 重要场景异常后不再执行其他异步任务现在重点展开标题里提到的核心知识点CompletableFuture异常后不再执行其他的异步任务。5.1 默认行为验证先看一段代码public class WorkflowExample { private static final ExecutorService executorService Executors.newFixedThreadPool(4); public static void main(String[] args) { CompletableFutureString future CompletableFuture .supplyAsync(() - step1(), executorService) .thenApplyAsync(result1 - step2(result1), executorService) .thenApplyAsync(result2 - step3(result2), executorService); try { future.join(); } catch (CompletionException e) { System.out.println(工作流异常终止 e.getCause().getMessage()); } } private static String step1() { System.out.println(step1 执行完成); return step1; } private static String step2(String input) { System.out.println(step2 执行开始); throw new RuntimeException(step2 模拟业务异常); } private static String step3(String input) { System.out.println(step3 执行完成); return step3; } }运行输出step1 执行完成 step2 执行开始step3不会被打印说明异常发生后后续的thenApplyAsync任务直接被跳过。最终future.join()抛出CompletionException外部可以看到异常信息。这是CompletableFuture链路默认的异常传播机制前序任务异常后续任务不会执行异常会沿着 CompletableFuture 链传递到最终结果。5.2 中间的 exceptionally 会改变行为接下来是容易踩坑的地方。如果在链路中间加一个exceptionally异常会被“拦截”后续任务会继续执行CompletableFutureString future CompletableFuture .supplyAsync(() - step1(), executorService) .thenApplyAsync(result1 - step2(result1), executorService) .exceptionally(ex - { System.out.println(异常被处理 ex.getMessage()); return 默认值; }) .thenApplyAsync(result3 - step3(result3), executorService);此时step3会执行因为exceptionally已经返回了一个默认值“默认值”CompletableFuture链认为异常已经恢复继续向下一步传递。这种设计在某些场景下是有用的。比如其中一个非关键依赖失败了你希望用一个默认值顶上让流程继续跑。但如果步step2是关键业务你希望异常发生后整个工作流停止那exceptionally就会破坏这个预期。所以规则要记清楚链中任意环节抛出异常后续thenApply/thenCompose不会执行。一旦出现exceptionally或handle并且成功返回异常将被视为已恢复后续任务会继续执行。如果想让异常一直向上传播不要在中间拦截或者拦截后重新抛出。5.3 异常隔离不要让一个任务的异常污染整条链有些场景下你既要顺序执行又要保证某个任务的异常不影响后续任务。此时不应该在链路中间插入exceptionally而应该把任务结果包装成Try或Either类型让异常作为数据传递。这里提供一个简单的结果包装类public class TaskResultT { private final T value; private final Throwable throwable; private TaskResult(T value, Throwable throwable) { this.value value; this.throwable throwable; } public static T TaskResultT success(T value) { return new TaskResult(value, null); } public static T TaskResultT failure(Throwable throwable) { return new TaskResult(null, throwable); } public boolean isSuccess() { return throwable null; } public T getValue() { return value; } public Throwable getThrowable() { return throwable; } }每个异步任务内部自己捕获异常把结果包装为TaskResultpublic CompletableFutureTaskResultStep2Output safeStep2(Step1Output input) { return CompletableFuture.supplyAsync(() - { try { return TaskResult.success(doStep2(input)); } catch (Throwable t) { return TaskResult.failure(t); } }, executorService); }这样即使step2抛异常TaskResult也会正常返回CompletableFuture链不会中断。下一步骤可以在拿到TaskResult后判断isSuccess()决定是继续还是提前返回。这种写法的好处是把异常从控制流中剥离出来变成数据流的一部分。工作流可以精确控制每个任务的失败处理策略而不是被CompletableFuture的异常传播规则牵着走。6. 真实场景订单状态流转工作流下面用一个订单状态流转的例子把上面的 API 串起来。需求如下步骤 A校验订单参数返回校验后的订单号。步骤 B查询订单详情返回订单实体。步骤 C根据订单详情生成对账单。步骤 D将对账单写入数据库。要求如果中间任何一步异常整个工作流停止不执行后续步骤。定义任务输入输出类型public record OrderId(String value) { } public record OrderDetail(String orderId, BigDecimal amount, String status) { } public record ReconciliationReport(String orderId, BigDecimal amount, String reportNo) { }工作流执行类Component public class OrderWorkflowService { private final ExecutorService workflowExecutor; private final OrderRepository orderRepository; private final ReconciliationReportGenerator reportGenerator; public OrderWorkflowService(OrderRepository orderRepository, ReconciliationReportGenerator reportGenerator) { this.orderRepository orderRepository; this.reportGenerator reportGenerator; this.workflowExecutor Executors.newFixedThreadPool(8, new ThreadFactory() { Override public Thread newThread(Runnable r) { Thread t new Thread(r, order-workflow- ThreadLocalRandom.current().nextInt(100)); t.setDaemon(true); return t; } }); } public CompletableFutureReconciliationReport executeWorkflow(OrderId orderId) { return CompletableFuture .supplyAsync(() - validateOrder(orderId), workflowExecutor) .thenApplyAsync(this::loadOrderDetail, workflowExecutor) .thenApplyAsync(this::generateReport, workflowExecutor) .thenApplyAsync(this::saveReport, workflowExecutor); } private OrderId validateOrder(OrderId orderId) { if (orderId null || orderId.value().isBlank()) { throw new IllegalArgumentException(订单号不能为空); } return orderId; } private OrderDetail loadOrderDetail(OrderId orderId) { OrderDetail detail orderRepository.findByOrderId(orderId.value()); if (detail null) { throw new IllegalStateException(订单不存在: orderId.value()); } return detail; } private ReconciliationReport generateReport(OrderDetail detail) { return reportGenerator.generate(detail); } private ReconciliationReport saveReport(ReconciliationReport report) { orderRepository.saveReport(report); return report; } public void shutdown() { workflowExecutor.shutdown(); } }调用方只需要拿到Future然后决定是阻塞等待还是异步回调OrderWorkflowService service applicationContext.getBean(OrderWorkflowService.class); CompletableFutureReconciliationReport reportFuture service.executeWorkflow(new OrderId(202506001)); reportFuture .thenAccept(report - System.out.println(工作流完成: report)) .exceptionally(ex - { System.err.println(工作流失败: ex.getMessage()); return null; });在这段代码里只要validateOrder、loadOrderDetail、generateReport、saveReport中任意一个抛出异常后面的步骤都不再执行。异常会传播到exceptionally回调中调用方可以感知并处理。7. 接口 API 与批量任务把工作流暴露成服务顺序工作流写完后通常会以两种形式对外提供服务REST API 和批量任务队列。7.1 封装为 REST API使用 Spring Boot 时可以写一个RestController接收请求内部提交异步工作流立即返回任务号然后通过轮询或回调通知结果。完整异步非阻塞写法需要配合DeferredResult或WebFlux这里给一个简单版本。RestController RequestMapping(/api/workflow/order) public class OrderWorkflowController { private final OrderWorkflowService workflowService; public OrderWorkflowController(OrderWorkflowService workflowService) { this.workflowService workflowService; } PostMapping(/async) public ResponseEntityString submitAsync(RequestBody OrderRequest request) { OrderId orderId new OrderId(request.getOrderId()); workflowService.executeWorkflow(orderId) .thenAccept(report - System.out.println(订单对账完成: report.reportNo())) .exceptionally(ex - { System.err.println(订单对账失败: ex.getMessage()); return null; }); return ResponseEntity.accepted().body(任务已提交结果将在后台异步处理); } PostMapping(/sync) public ResponseEntityReconciliationReport submitSync(RequestBody OrderRequest request) { OrderId orderId new OrderId(request.getOrderId()); ReconciliationReport report workflowService.executeWorkflow(orderId).join(); return ResponseEntity.ok(report); } public static class OrderRequest { private String orderId; public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId orderId; } } }/async接口适合前端不关心同步结果的场景提交后直接返回 202后台通过日志或消息通知跟进。/sync接口适合内部系统调用调用方需要立即拿到结果。7.2 批量任务流水线如果需要批量处理大量订单可以结合ExecutorService和CompletableFuture.allOf实现批量提交。public ListReportResult executeBatch(ListOrderId orderIds) { ListCompletableFutureReportResult futures orderIds.stream() .map(orderId - workflowService.executeWorkflow(orderId) .thenApply(report - ReportResult.success(orderId.value(), report.reportNo())) .exceptionally(ex - ReportResult.failure(orderId.value(), ex.getMessage()))) .toList(); CompletableFutureVoid allDone CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); allDone.join(); return futures.stream() .map(CompletableFuture::join) .toList(); }上面这段代码有几个关键点每个订单独立提交异步工作流。每个工作流的异常在exceptionally中被打包成ReportResult.failure不会影响其他订单的执行结果。这是批量任务必须做的异常隔离。allOf等待所有任务完成然后逐个join收集结果。从线程池设计角度看批量任务的线程数不能太大否则会打满数据库连接池。建议根据服务配置的数据库连接池大小和下游接口的吞吐能力动态调整工作流线程池的线程数。7.3 任务队列存储以上批量处理都是内存中的方案一旦服务重启所有未完成的任务都会丢失。对于生产环境更稳妥的做法是把任务状态写入数据库或 Redis使用消息队列如 RocketMQ、Kafka 驱动每步执行。这里给出一种任务状态存储表的结构供参考字段类型说明task_idVARCHAR(64)任务唯一 IDorder_idVARCHAR(64)业务订单号current_stepINT当前执行到的步骤statusVARCHAR(32)PENDING / RUNNING / SUCCESS / FAILEDerror_messageTEXT失败原因retry_countINT重试次数create_timeDATETIME创建时间update_timeDATETIME更新时间每执行一个步骤先查询当前状态再执行对应步骤最后更新current_step和status。这样做的好处是任务中断后可以从断点恢复而不是从头开始。8. 资源占用与性能观察异步工作流的资源占用主要体现在线程池、内存和 CPU 三个方面。8.1 线程池观察线程池的状态可以通过ThreadPoolExecutor暴露的指标观察ThreadPoolExecutor executor (ThreadPoolExecutor) workflowExecutor; System.out.println(核心线程数: executor.getCorePoolSize()); System.out.println(活跃线程数: executor.getActiveCount()); System.out.println(队列中任务数: executor.getQueue().size()); System.out.println(已完成任务数: executor.getCompletedTaskCount());如果getQueue().size()持续增长说明任务生产速度大于消费速度需要评估是否增加线程数或者限制入口流量。8.2 超时控制CompletableFuture默认没有强制超时机制。JDK 9 开始提供了orTimeout和completeOnTimeout。建议在调用外部依赖时设置超时CompletableFutureOrderDetail future CompletableFuture .supplyAsync(() - orderRepository.findByOrderId(orderId), executor) .orTimeout(3, TimeUnit.SECONDS) .exceptionally(ex - { // 超时或异常时返回默认值 return new OrderDetail(orderId, BigDecimal.ZERO, TIMEOUT); });这里要小心orTimeout触发后任务本身可能还在跑异常回调会立即返回。如果下游依赖已经挂掉超时设置能避免调用方无限期等待。8.3 降低无效资源占用一个常见的性能问题是CompletableFuture的链路很长但线程池线程数太少任务在线程池队列里排队导致链路整体延迟变大。观察工具可以用 Arthas 的thread命令查看线程状态或者用 JProfiler 导出线程 dump查看每个工作流线程停留在哪个阶段。如果发现大部分任务都在等待 IO可以把 IO 密集型任务和 CPU 密集型任务拆到不同的线程池中避免互相挤占。9. 常见问题与排查方法从过往经验看顺序工作流异步执行最容易遇到下面几类问题。问题现象可能原因排查方式解决方案异常后后续任务仍然执行链路中存在exceptionally/handle并返回了默认值检查链路中所有异常处理方法将exceptionally放在链路末尾或在业务代码中重新抛出异常join()抛出CompletionException看不到原始错误CompletableFuture包装了异常使用e.getCause()获取真实异常打印e.getCause().getMessage()或完整堆栈任务提交后没有任何输出线程池队列积压任务等待执行观察线程池活跃线程数和队列大小调整线程池参数或扩大线程数批量任务中一个任务失败导致其他任务不执行使用allOf().join()且某个 future 未隔离异常检查每个 future 是否有exceptionally在每个任务内捕获异常并包装为结果对象服务重启后任务丢失没有持久化任务状态检查是否有数据库或 Redis 中的任务表引入任务状态表支持断点续跑CPU 飙升线程数过多或任务内有死循环JProfiler 导出 CPU 采样限制线程数优化任务逻辑数据库连接池耗尽每个任务都占用一个连接并长时间等待查看数据库连接池活跃连接数降低线程池大小或使用连接池隔离最核心的一条排查经验先确认异常到底在哪一层被吞掉再看链路是否偏离预期。推荐在所有异步任务入口加日志例如private String step2(String input) { System.out.println(step2 开始执行输入: input); try { // 业务逻辑 } catch (Exception ex) { System.err.println(step2 异常: ex.getMessage(), ex); throw ex; } }在高并发场景可以把System.out替换为 SLF4J 日志并把任务 ID 放入 MDC这样排查问题时可以按任务 ID 串起整条链路的日志。10. 最佳实践与使用建议到这一步你已经能把顺序工作流用CompletableFuture异步跑起来。下面是几条工程化建议直接影响线上稳定性。10.1 线程池必须单独定义不要用Executors.newFixedThreadPool()创建无界队列线程池更不要直接把ForkJoinPool.commonPool()作为工作流线程池。推荐使用ThreadPoolExecutor并手动指定队列类型和拒绝策略ThreadPoolExecutor executor new ThreadPoolExecutor( 8, 16, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue(200), new ThreadFactory() { Override public Thread newThread(Runnable r) { Thread t new Thread(r, workflow-pool- ThreadLocalRandom.current().nextInt(100)); t.setDaemon(true); return t; } }, new ThreadPoolExecutor.CallerRunsPolicy() );CallerRunsPolicy在队列满时不会抛弃任务而是由调用方线程执行牺牲同步调用方的性能换取任务不丢失。10.2 异常策略要提前定义工作流每个步骤的异常处理策略应该在设计阶段确定关键步骤异常整个工作流终止。非关键步骤异常允许降级或重试。批量任务中单条失败不影响其他条。建议把策略抽象成一个枚举类public enum StepErrorPolicy { STOP_ON_FAILURE, CONTINUE_WITH_DEFAULT, RETRY_AND_STOP }然后在每个步骤的包装方法中根据策略决定是抛出异常、返回默认值还是重试。10.3 异步任务要设置超时远程调用、数据库访问、文件 IO 都可能卡住。CompletableFuture的orTimeout是 JDK 9 的能力如果你必须兼容 JDK 8可以封装一个带超时的执行方法public T CompletableFutureT executeWithTimeout( SupplierT task, long timeout, TimeUnit unit) { CompletableFutureT future CompletableFuture.supplyAsync(task, executor); // JDK 8 下的超时模拟在另一个线程里等待完成 ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); scheduler.schedule(() - { if (!future.isDone()) { future.completeExceptionally(new TimeoutException(任务执行超时)); } }, timeout, unit); return future; }JDK 8 没有原生orTimeout但可以用手动调度实现类似效果。生产环境务必确认超时兜底否则一个下游接口挂掉会拖垮整个工作流。10.4 可观测性要起监控建议为工作流的每个步骤埋点统计执行时间和失败次数。可以用 Micrometer 发布指标也可以简单打日志。关键指标每个步骤的平均耗时和 P99 耗时。工作流整体成功率。线程池活跃线程数和队列长度。有了这些数据才能判断到底是哪一步拖慢了整个流程哪些步骤适合并行哪些步骤必须串行。11. 总结与下一步这次的内容核心是顺序工作流的异步执行以及CompletableFuture异常后不再执行其他异步任务这个问题。最值得先动手验证的点是写一段 3 到 4 步的串联工作流在第 2 步故意抛异常观察后续步骤是否执行。你会发现CompletableFuture的默认行为是正确的后续任务会跳过。如果你发现后续任务仍然执行回头看链路里是不是有exceptionally拦截了异常。最容易踩的坑有两个在链路中间用exceptionally恢复异常导致后续步骤继续执行。批量任务里没有做异常隔离一个任务失败导致allOf().join()抛出异常。后续可以继续扩展的方向包括用虚拟线程替换普通线程池在 JDK 21 上编写顺序工作流代码更简单、资源占用更低。将工作流步骤配置化比如用 JSON 定义步骤顺序、重试次数、失败策略实现“工作流即配置”。引入可视化工作流引擎比如 Flowable、Camunda把异步编排从代码层面提升到流程引擎层面。如果你已经在用CompletableFuture做异步任务编排可以先把线程池指标和异常策略补齐再考虑是否引入重量级工作流引擎。顺序工作流本身并不复杂复杂的是异常控制、批量隔离和可观测性。这三件事做好生产环境基本不会出大问题。