1. 为什么你的异步流总在半夜炸掉
CompletableFuture 的组合与异常处理，说白了就是解决一件事：多个异步任务怎么拼起来、拼完之后出错怎么办。它适合已经会写supplyAsync、thenApply，但一遇到thenCompose、allOf、exceptionally就心里没底的后端同学。我见过太多线上事故，不是单点接口挂了，而是某个异步分支抛了异常没人接，整条链静默失败，日志里干干净净，监控上风平浪静，直到用户投诉才发现数据没落库。
这篇不讲概念复读，直接给你一套能跑的异步流骨架：多任务组合用thenCompose/thenCombine/allOf，异常处理用exceptionally/handle/whenComplete，再配一个统一的模型调用通道，把散落在各处的 Key 收敛成一份配置。这样你排查问题时只需要看一个入口，而不是在五个服务的配置文件里翻。
核心检索词先摆出来：CompletableFuture 组合、异步流、异常处理、统一 Key。适合谁？写 Java 后端、做接口聚合、搞订单编排、需要并行调多个下游服务的同学。读完你能拿到三段东西：可复制的异步流骨架代码、TaoToken 的配置片段、以及异常分支的验证动作和预期输出。
2. 前置准备：用 TaoToken 统一 Key 收敛调用入口
异步流最怕的不是逻辑复杂，而是依赖分散。你调三个模型服务，就有三套地址、三个 Key、三种超时策略，异常处理写起来像打补丁。我的做法是先把调用通道统一，所有异步任务走同一个 API 入口，Key 只维护一份。
TaoToken 在这里扮演的角色就是统一通道：官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api 。你注册后在控制台生成 Key，然后把它写进配置，代码里只读配置不硬编码。
先看配置骨架。如果你用 Java 项目，习惯config.toml风格，可以这样写：
# config.toml [taotoken] base_url = "https://taotoken.net/api" api_key = "sk-你的Key" default_model = "claude-sonnet" connect_timeout_ms = 1000 read_timeout_ms = 2000如果你更习惯 JSON 配置，比如某些工具链读settings.json：
{ "taotoken": { "baseUrl": "https://taotoken.net/api", "apiKey": "sk-你的Key", "defaultModel": "claude-sonnet", "timeoutMs": 2000 } }Key 的获取入口在控制台的 API Keys 页面，地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。生成之后别写进代码，用环境变量或配置中心注入。我试过把 Key 直接塞进application.yml提交到仓库，结果被扫描工具告警，虽然没出事但流程上很被动。
注意：配置里的超时时间要和后面 CompletableFuture 的
orTimeout对齐，否则会出现「HTTP 层还没超时，Future 层已经判失败」的错位，排查起来很绕。
统一通道之后，你的异步任务只需要关心「调什么模型、传什么参数」，不用再关心「连哪个地址、用哪个 Key」。这一步是后面所有组合和异常处理的地基。
3. 可复制的 CompletableFuture 异步流骨架
下面这段代码是完整可跑的骨架，我把它拆成三层：线程池、组合逻辑、异常兜底。你直接复制改业务方法名就能用。
3.1 线程池与基础封装
import java.util.concurrent.*; import java.util.*; import java.util.stream.Collectors; public class AsyncFlowSkeleton { // 独立 IO 线程池，避免和业务线程池互相拖累 private static final ExecutorService IO_POOL = new ThreadPoolExecutor( 8, 16, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200), new ThreadFactory() { private int i = 0; public Thread newThread(Runnable r) { Thread t = new Thread(r, "async-io-" + (i++)); t.setDaemon(true); return t; } }, new ThreadPoolExecutor.CallerRunsPolicy() ); // 通用：把多个 Future 收集成 List，带独立异常兜底 public static <T> CompletableFuture<List<T>> allOfList( List<CompletableFuture<T>> futures, T fallback) { List<CompletableFuture<T>> safe = futures.stream() .map(f -> f.exceptionally(ex -> { System.out.println("[warn] 任务失败，使用兜底值: " + ex.getMessage()); return fallback; })) .collect(Collectors.toList()); return CompletableFuture.allOf(safe.toArray(new CompletableFuture[0])) .thenApply(v -> safe.stream() .map(CompletableFuture::join) .collect(Collectors.toList())); } }这里的关键点是exceptionally放在allOf之前。如果你先allOf再处理异常，只要有一个任务失败，整个allOf就直接异常完成，其他成功的结果你拿不到。先给每个任务套一层兜底，allOf就永远不会因为单个失败而崩。
3.2 thenCompose 级联：后一个任务依赖前一个结果
public CompletableFuture<String> cascadingFlow(String userId) { // 第一级：拿用户信息 CompletableFuture<String> userFuture = CompletableFuture .supplyAsync(() -> fetchUser(userId), IO_POOL) .orTimeout(1, TimeUnit.SECONDS) .exceptionally(ex -> "anonymous-user"); // 第二级：依赖用户结果，再拿订单 CompletableFuture<String> orderFuture = userFuture.thenCompose(user -> CompletableFuture.supplyAsync(() -> fetchOrders(user), IO_POOL) .orTimeout(2, TimeUnit.SECONDS) .exceptionally(ex -> "[]") ); // 第三级：依赖用户结果，并行拿优惠券 CompletableFuture<String> couponFuture = userFuture.thenCompose(user -> CompletableFuture.supplyAsync(() -> fetchCoupons(user), IO_POOL) .orTimeout(1, TimeUnit.SECONDS) .exceptionally(ex -> "[]") ); // 合并订单和优惠券 return orderFuture.thenCombine(couponFuture, (orders, coupons) -> buildResponse(orders, coupons)); }thenCompose和thenApply的区别要记牢：thenApply的入参是普通值，返回普通值；thenCompose的入参是普通值，返回的是CompletableFuture。如果你在thenApply里返回一个 Future，你会得到CompletableFuture<CompletableFuture<T>>，嵌套两层，后面join的时候很痛苦。级联调用一律用thenCompose。
3.3 thenCombine 并行合并：两个独立任务
public CompletableFuture<String> parallelMerge(String userId) { CompletableFuture<String> profileFuture = CompletableFuture .supplyAsync(() -> fetchProfile(userId), IO_POOL) .orTimeout(1, TimeUnit.SECONDS) .exceptionally(ex -> "{}"); CompletableFuture<String> statsFuture = CompletableFuture .supplyAsync(() -> fetchStats(userId), IO_POOL) .orTimeout(1, TimeUnit.SECONDS) .exceptionally(ex -> "{}"); return profileFuture.thenCombine(statsFuture, (profile, stats) -> mergeJson(profile, stats)); }thenCombine适合两个任务互不依赖、但结果要合并的场景。串行执行是t1 + t2，并行是max(t1, t2)，接口聚合场景下这个差距很可观。
3.4 异常处理三件套的定位
exceptionally只处理异常，返回兜底值，异常被阻断。handle同时处理成功和失败，可以改变结果类型。whenComplete只做副作用，比如打日志、埋点，不改变结果，异常继续往下传。
public CompletableFuture<String> withObservability(String userId) { return CompletableFuture .supplyAsync(() -> riskyCall(userId), IO_POOL) .orTimeout(2, TimeUnit.SECONDS) .handle((result, ex) -> { if (ex != null) { System.out.println("[handle] 捕获异常: " + ex.getMessage()); return "fallback"; } return result; }) .whenComplete((r, ex) -> { // 这里 ex 永远是 null，因为 handle 已经恢复了 System.out.println("[whenComplete] 最终结果: " + r); }); }顺序很重要：handle在前，whenComplete在后，whenComplete看到的就是已经恢复后的正常结果。如果你把whenComplete放在handle前面，它能看到原始异常，适合做告警。
4. 验证请求与预期输出
光看代码不算数，得跑起来验证。下面给三个验证动作，你照着做能确认异常分支真的生效。
4.1 验证 allOf 的独立兜底
构造三个任务，第二个故意抛异常：
List<CompletableFuture<String>> futures = Arrays.asList( CompletableFuture.supplyAsync(() -> "task-1-ok", IO_POOL), CompletableFuture.supplyAsync(() -> { throw new RuntimeException("task-2-boom"); }, IO_POOL), CompletableFuture.supplyAsync(() -> "task-3-ok", IO_POOL) ); List<String> results = allOfList(futures, "fallback").join(); System.out.println(results);预期输出：
[warn] 任务失败，使用兜底值: java.lang.RuntimeException: task-2-boom [task-1-ok, fallback, task-3-ok]如果你看到的是抛异常而不是这个列表，说明exceptionally的位置放错了，检查是不是写在了allOf之后。
4.2 验证 orTimeout 触发
CompletableFuture<String> slow = CompletableFuture .supplyAsync(() -> { try { Thread.sleep(3000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "slow-done"; }, IO_POOL) .orTimeout(1, TimeUnit.SECONDS) .exceptionally(ex -> "timeout-fallback"); System.out.println(slow.join());预期输出：
timeout-fallback注意orTimeout是 JDK 9+ 的 API，JDK 8 项目得自己用ScheduledExecutorService实现，思路是起一个定时任务，到点如果原 Future 没完成就completeExceptionally。
4.3 验证统一 Key 通道连通
配置写好后，先用一个最小请求确认通道可用。你可以用 curl 直接打 API：
curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "claude-sonnet", "messages": [{"role": "user", "content": "ping"}] }'预期返回一个包含choices字段的 JSON。如果返回 401，检查 Key 是否带上了Bearer前缀；如果返回超时，检查connect_timeout_ms是不是设得太小。通道通了，再把这段逻辑包进supplyAsync里，异步流才有意义。
想先在网页上确认模型能正常对话，可以走模型对话入口 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，发一句话看响应，确认账号和模型都没问题，再回到代码里调。
5. 本篇常见错排查
5.1 异常被静默吞掉
最常见的坑：链尾没有exceptionally或handle，异常就停在 Future 内部，join的时候才抛CompletionException。如果你连join都没调，异常就彻底消失了。排查方法：给每个异步链的末尾强制加一个whenComplete打日志，先让异常可见。
5.2 allOf 拿不到结果
CompletableFuture.allOf返回的是CompletableFuture<Void>，它只告诉你「都完成了」，不给你结果。你必须手动从各个子 Future 里join。而且join之前要确认子 Future 已经完成，否则会阻塞。正确姿势是allOf(...).thenApply(v -> futures.stream().map(CompletableFuture::join)...)，在thenApply里join不会阻塞，因为此时所有子任务都完成了。
5.3 thenCompose 和 thenApply 混用
在thenApply里返回CompletableFuture，会得到嵌套 Future，后面join两次才能拿到值。看到CompletableFuture<CompletableFuture<X>>这种类型，立刻改成thenCompose。
5.4 线程池打满导致 CallerRunsPolicy 反压
上面骨架里用了CallerRunsPolicy，队列满了之后由调用线程执行任务。这在异步流里是把双刃剑：好处是不会丢任务，坏处是可能阻塞主线程。如果你的异步流嵌套很深，建议换成AbortPolicy加显式降级，或者把队列调大并监控活跃线程数。
5.5 超时时间层层叠加
HTTP 客户端超时 2 秒，orTimeout设 3 秒，上游网关超时 1 秒。结果网关先断，你的 Future 还在跑，资源白占。原则是外层超时小于内层，逐层收敛。统一 Key 通道的read_timeout_ms要和orTimeout对齐，别一个 2 秒一个 5 秒。
6. 把异步流接进长期编码工作流
骨架跑通之后，下一步是把它变成日常开发的一部分。如果你经常写这类异步编排代码，可以考虑用 Coding Plan 把模型调用、代码补全、异常排查串成一条工作流，入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。它适合长期写 Java 后端、需要反复调试异步链的场景，Key 和通道还是走同一套配置，不用再单独维护。
接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，里面有完整的参数说明和错误码对照。遇到 429 限流或者 5xx 服务端错误，先查文档里的错误码表，再决定是重试还是降级。
最后留一个我踩过的坑：whenComplete里不要做耗时操作，它运行在完成线程上，可能阻塞整个链的后续任务。打日志、埋点可以，发 HTTP 请求不行。要发请求就再起一个supplyAsync，把副作用异步化。异步流的可观测性靠日志和埋点，但别让观测本身变成新的阻塞点。
