1. 从一次消息推送事故说起我在做某内部运营系统的消息推送模块时遇到过一个很典型的问题系统要给员工批量发送企业微信应用消息高峰期一次性要推几千条。最初的实现很直接用一个循环去逐个调用企业微信的发送接口每次调用都同步等待HTTP响应。结果就是单线程串行发送一秒钟只能发几条几十条不等批量推送任务经常要跑十几分钟甚至更久而且一旦中间某条消息发送失败整个任务的进度和状态管理就很混乱。后来改用多线程并发发送确实快了不少但新的问题又来了企业微信接口有频率限制盲目把几百个线程同时打上去直接触发频控大量请求返回错误码账号甚至被短暂限制调用。而且代码里用裸线程管理并发线程资源难以控制失败重试、超时处理、结果汇总这些逻辑写起来非常别扭。那段时间正好在系统性地看Java并发相关的技术资料注意到了CompletableFuture这个类。它和我之前用的Future最大的区别是支持异步编排——可以把多个异步任务串成链条、合并结果、并行执行、异常兜底写出来的代码像在描述业务流程而不是陷在回调地狱里。我借着改造企业微信批量消息发送这个场景把CompletableFuture用得比较彻底。这篇文章就围绕这个实战改造展开记录下异步编排的设计思路、核心代码、踩坑经验希望能给正在处理同类批量推送问题的开发者一些参考。如果你还没用过CompletableFuture或者用过但只停留在supplyAsync加join这种简单组合的阶段这篇文章会比较适合你。我会把编排思路、线程池选型、频控处理和失败恢复这些关键环节都拆开讲清楚。2. 批量消息发送为什么需要异步编排2.1 同步串行发送的性能瓶颈先看最原始的串行方案大概长什么样// 串行发送伪代码 for (Message message : messageList) { SendResult result wecomClient.send(message); handleResult(result); }这段代码看起来没毛病逻辑也简单清晰。但放到真实场景里性能问题会非常突出。企业微信发送接口是走HTTPS的一次请求从建立连接到服务端处理、返回响应乐观估计也要几十毫秒。如果按平均80毫秒算串行发送1000条消息就是80秒这还只是纯请求耗时没算上消息内容处理、日志记录、结果落库的时间。如果是几千条甚至上万条任务耗时就是几十分钟量级。这在运营场景里是难以接受的——业务方可能着急让全员马上收到通知却要等大半个小时体验非常糟糕。串行方案的另一个问题是对单条消息的失败缺乏隔离能力。一条消息发送超时默认的超时时间可能长达几十秒后面的所有消息都要排队等着。要是连续几条都超时整个发送队列就堵死了。2.2 多线程并发为什么还不够串行不行那就并发。很多人第一步会想到用线程池加Future把消息列表拆成多个子任务并行提交ExecutorService executor Executors.newFixedThreadPool(10); ListFutureSendResult futures new ArrayList(); for (Message message : messageList) { futures.add(executor.submit(() - wecomClient.send(message))); } for (FutureSendResult future : futures) { SendResult result future.get(); handleResult(result); }这个方案发送速度确实上来了10个线程并发就是近10倍的提升。但很快就遇到几个实际问题第一是频控问题。企业微信对应用消息发送有频率限制超出后会返回特定错误码。并发线程数开大了容易触发限制开小了性能提升又有限而且这个限制并不是简单算总条数和调用频率也有关。用裸线程池加Future的方式缺少一个优雅的“限速”编排手段。第二是异常处理和补偿逻辑很分散。future.get()会抛出ExecutionException你得在外层catch然后逐个处理失败的条目。如果发送失败后要重试重试逻辑会因为原始任务已经执行完而变得复杂你得重新提交新任务。第三是编排能力缺失。真实场景里发送消息之前需要先获取access_tokenaccess_token有过期时间而且获取token本身也是一个耗时操作。多线程并发发送时必须保证token只获取一次同时所有发送任务都要等token就绪后才能开始。用Future实现这种“先等一个任务完成、再让所有任务同时开始”的逻辑不太顺手。2.3 CompletableFuture带来的编排思路转变CompletableFuture换个角度看是把异步任务描述成一张依赖图。你不需要关心“哪个线程在什么时候执行哪一步”只需要描述“这一步依赖上一步的结果”以及“哪些步骤可以同时进行”。拿消息发送来说整个流程可以拆成几段获取access_token全局只执行一次使用token调用发送接口大量并行任务每个发送任务分别处理结果成功日志、失败记录所有任务完成后统一汇总状态用CompletableFuture描述这个流程每一步都是一个CompletableFuture节点节点之间通过thenApply、thenCompose、allOf、exceptionally这些方法组合起来。代码读起来像流水线执行起来却是分叉并行的。这就是异步编排的核心思想让代码结构贴合任务依赖关系而不是贴合线程模型。这也正是标题里“异步编排”四个字的含义。它解决的不只是“跑得快”而是“跑得稳、跑得可控、失败可恢复”。3. 异步编排在批量发送中的具体应用3.1 任务分解三个阶段的依赖关系企业微信批量消息发送的完整流程我在设计时拆成了三个阶段第一阶段准备阶段。获取access_token在应用配置中读取消息模板内容按发送范围内的成员列表组装消息实体。这个阶段是后续所有任务的前置条件。access_token的获取比较特别它自带缓存逻辑缓存没失效就直接复用失效了才真正调接口刷新。这个阶段的结果是一份只读的上下文对象。第二阶段发送阶段。把消息列表按一定规则分组每组作为一个独立异步任务提交执行。每个任务内部拿到上下文中的access_token循环发送组内的消息。这里的分组不是随便分的要结合企业微信的频控阈值来设计既要让每组内消息数量适度又要保证并发组数不会冲击频控。第三阶段汇总阶段。所有发送任务完成后把每个任务的执行结果汇总起来统计成功条数、失败条数、失败明细更新任务状态。如果失败率达到阈值触发补偿发送流程。这三个阶段的依赖关系是阶段2依赖阶段1的结果阶段3依赖阶段2全部完成。但阶段2内部的多个发送任务之间互不依赖。这种结构非常适合用CompletableFuture来编排。3.2 核心代码骨架CompletableFuture的组合方式先定义上下文对象它贯穿整个异步流程public class SendContext { private final String accessToken; private final ListMessage messages; private final AtomicInteger successCount new AtomicInteger(0); private final AtomicInteger failCount new AtomicInteger(0); private final ConcurrentLinkedQueueFailedMessage failedMessages new ConcurrentLinkedQueue(); public SendContext(String accessToken, ListMessage messages) { this.accessToken accessToken; this.messages messages; } }需要说明的是我的习惯是让整个发送流程阻塞等待最终结果但内部所有子任务都是异步并发的。这样既保留了异步编排的性能收益又让调用方的业务逻辑是同步连贯的。如果你希望调用方也完全异步可以把末尾的join换成注册回调。接着是核心编排入口方法public SendSummary sendBatch(ListMessage messages) { // 阶段1异步获取access_token内部有缓存逻辑 CompletableFutureString tokenFuture CompletableFuture.supplyAsync( this::getAccessToken, tokenExecutor); // 阶段2token就绪后拆分成多个发送任务并行执行 CompletableFutureVoid allSends tokenFuture.thenCompose(token - { SendContext context new SendContext(token, messages); ListCompletableFutureVoid sendFutures partitionMessages(messages) .stream() .map(batch - CompletableFuture.runAsync( () - sendBatchInternal(context, batch), sendExecutor)) .collect(Collectors.toList()); return CompletableFuture.allOf(sendFutures.toArray(new CompletableFuture[0])); }); // 阶段3全部发送结束后汇总结果 CompletableFutureSendSummary summaryFuture allSends .thenApply(v - buildSummary(context)) .exceptionally(ex - buildFailedSummary(ex)); return summaryFuture.join(); }这段代码的巧妙之处在于几个关键点tokenFuture.thenCompose实现了“等token拿到再开始发送”的依赖关系。thenCompose和thenApply的区别我用一个类比解释。如果你有一个CompletableFutureString想基于字符串进一步得到一个CompletableFutureInteger用thenApply嵌套结果是CompletableFutureCompletableFutureInteger相当于盒子套盒子。用thenCompose就直接展开成CompletableFutureInteger一步到位。这正是“先拿token再发消息”需要的token结果出来了下一步要开启新的异步任务发送而不是在同一个线程里同步处理。partitionMessages负责分批。每个批次是一个独立的异步子任务runAsync提交到发送线程池。发送线程池的线程数不是越多越好这个后面专门说。allOf是整条链路的汇合点。它返回的CompletableFutureVoid在所有子任务完成后才完成。注意allOf本身不处理异常任何一个子任务异常它都会以异常状态完成。所以每个发送子任务内部必须有异常兜底不能让异常逃逸出来破坏整个编排链。exceptionally是最后一道防线它确保即使发送链路整体出现极端情况比如token获取失败导致后续全部无法执行也能返回一个标记失败状态的SendSummary而不是让调用方直接收到CompletionException。这种编写方式最大的好处是你看到的代码顺序就是业务流程顺序阶段1完成后自动进入阶段2阶段2全部完成后自动汇总。线程切换的细节被框架消化掉了开发者只需要关注任务本身的逻辑。3.3 如何分批频控约束下的任务切分分批策略看起来是个小问题实际是性能和稳定性之间的平衡点。分得不够细一批消息太多单任务执行时间过长一旦该批中途失败重试成本很高。分得太细任务数量膨胀线程池频繁切换上下文反而降低吞吐还可能因为并发请求过多触发频控。企业微信的频控规则结合我自己的实测经验来看应用消息的发送不是看单一接口的每秒调用量那么简单它会综合账号维度的整体调用频次、内容维度的消息条数来评估。我在改造中采用的策略是以“每批50条最多同时运行10批”作为基准。为什么是50条单个发送接口一次调用也可以支持消息数组形式但携带大量非文本消息图片、卡片、文件等时请求体较大一次调用耗时明显上升。50条可以控制单次请求体量适中同时保证单批任务在一两秒内结束可控性强。为什么最多10批假设每批50条同时发10批就是500条并发向企业微信写数据。我在测试环境逐步加压后发现这个量级相对安全不会触发频控同时发送线程池线程数设置在10到16之间比较合理。分批的实际逻辑private ListListMessage partitionMessages(ListMessage messages) { ListListMessage partitions new ArrayList(); int batchSize 50; for (int i 0; i messages.size(); i batchSize) { int end Math.min(i batchSize, messages.size()); partitions.add(new ArrayList(messages.subList(i, end))); } return partitions; }这是一种静态分批法简单可靠。但如果你希望更动态地控制并发可以用信号量配合线程池做动态限流。我的做法比较务实先静态分批在任务内部再通过一个Semaphore控制每秒实际发出的请求量双保险。发送线程池我用了单独的实例而不是默认的ForkJoinPool.commonPool()。原因很简单默认公共池被整个JVM共享很多框架代码也会用它一旦发送任务阻塞可能影响别的不相关功能。而且公共池的线程数默认是处理器核心数减一对于IO密集型的HTTP调用来说这个线程数远远不够无法最大化吞吐。自定义线程池参数我放在后面“避坑经验”里细说。3.4 发送任务内部的try-catch策略每个发送子任务内部的容错设计是整个方案里容易被忽视但非常重要的一环。我先明确一个设计原则子任务内部的单条消息异常不应该让整个批次失败更不能让异常冒泡到allOf层面。一旦allOf因为某个子任务异常导致整条链以异常状态完成其他还在正常发送的子任务的结果就不好汇总了。所以我在runAsync内部一律用局部变量保存异常信息代码模式如下private void sendBatchInternal(SendContext context, ListMessage batch) { ListFailedMessage localFailed new ArrayList(); for (Message message : batch) { try { SendResult result wecomClient.send(context.getAccessToken(), message); if (result.isSuccess()) { context.getSuccessCount().incrementAndGet(); } else { handleFailure(context, message, result.getErrorMsg(), localFailed); } } catch (Exception ex) { handleFailure(context, message, ex.getMessage(), localFailed); } } if (!localFailed.isEmpty()) { // 标记批次局部失败但不抛出异常等待后续补偿逻辑 context.getFailedMessages().addAll(localFailed); } }但光有局部try-catch还不够。如果某条消息发送失败是因为access_token过期那整批消息都会陆续失败。我在handleFailure里做了个判断如果失败原因是指定的token失效错误码就触发token强制刷新并重试当前消息一次。注意这里为了简单重试只做一次否则可能陷入死循环。这就是典型的“区分可恢复异常和不可恢复异常”的思路。Compensating重试、局部失败、整链兜底这三层防护加起来才是一个生产级批量发送任务该有的容错姿态。4. 线程池选型与核心参数设计4.1 为什么不用默认的commonPool很多刚接触CompletableFuture的人会直接使用supplyAsync(() - doSomething())不带线程池参数。这样默认用的是ForkJoinPool.commonPool()。小体量任务可能看不出问题但放到批量消息发送这种每天要跑很多次的场景里坑就来了第一公共池的线程数是根据CPU核数算出来的通常很少。而企业微信发送这种任务本质是IO密集型大部分时间在等待HTTP响应线程大部分时间处于阻塞状态。要提升吞吐必须让更多线程处于“可运行”状态也就是增加线程数。公共池的线程数设置根本不够用。第二公共池是全局共享的。如果发送任务中出现了某个外部依赖的慢调用占用了公共池里的线程同期执行的其他异步任务都会被拖慢。我在生产环境遇到过因为一次内存缓存加载超时导致公共池线程被占满其他模块的异步任务大面积变慢的情况。自那以后所有异步任务我都尽量使用专用线程池隔离影响面。4.2 线程池参数怎么定我的发送线程池配置经验值如下private static final int SEND_CORE_POOL_SIZE 8; private static final int SEND_MAX_POOL_SIZE 16; private static final int SEND_QUEUE_CAPACITY 2000; private ThreadPoolExecutor createSendExecutor() { return new ThreadPoolExecutor( SEND_CORE_POOL_SIZE, SEND_MAX_POOL_SIZE, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue(SEND_QUEUE_CAPACITY), new NamedThreadFactory(wecom-send), new ThreadPoolExecutor.CallerRunsPolicy() ); }参数解读核心线程数8最大线程数16。因为任务是IO密集型的单个线程大部分时间在等待网络响应所以线程数可以比CPU核数多几倍。8到16是我根据并发测试结果选的。再往上加企业微信服务端的响应时间就会明显变长反而浪费资源。队列容量2000这个值很关键。如果队列太小批次数超过2000后新任务就会触发拒绝策略导致部分消息发送被直接丢弃。如果队列太大极端情况下内存压力大。2000配合每批50条意味着理论上可以缓冲10万条消息的发送任务足够覆盖日常峰值。拒绝策略用了CallerRunsPolicy。这个策略的意思是当线程池和队列都满了新提交的任务不丢弃而是由提交任务的线程自己执行。这样做的效果是不会丢任务同时提交方线程因为要执行任务而被“拖住”天然形成背压降低提交速度。在批量发送场景里宁可慢一点也不能丢消息。token获取线程池单独配只要2个线程就够了。因为token获取是低频操作有缓存兜底两个线程足够保证同时只有一个线程真正在调refresh接口另一个可以用来应急。线程池关闭的时机也很重要。批量任务全部完成后不要立刻shutdown线程池因为线程池是复用的下次任务还要用。我是在Spring容器销毁时统一关闭平时线程池核心线程保持存活。4.3 线程池和CompletableFuture的配合细节CompletableFuture的每个组合方法都有带线程池参数的重载版本。要不要每个方法都显式传线程池其实不用。我的用法是supplyAsync、runAsync这种真正开启新异步任务的入口方法必须显式传线程池。而thenApply、thenCompose、exceptionally这些后续操作默认会在前一个任务完成的线程上执行不需要额外传线程池。这个特性有个好处thenCompose里的发送子任务提交虽然发生在token获取完成的那个线程上但子任务本身异步执行不会阻塞token线程。有一种情况需要注意。allOf返回的future在全部子任务完成时触发后续的thenApply如果子任务数量很大allOf内部也是靠回调链管理的回调执行线程是最后一个完成的子任务线程。这个线程可能属于发送线程池也可能属于业务线程无妨因为汇总逻辑执行很快。5. 避坑经验与排查实录5.1 CompletableFuture的异常传递坑这个坑我印象深刻。第一次用CompletableFuture改造时我在thenApply里写了可能抛异常的逻辑以为外层catch能捕获到结果发现异常变成了CompletionException包了一层。这是因为CompletableFuture内部的异常包装机制任何一个环节抛出的异常最终都会包装成CompletionException传递给下游。这个机制有好处也有坏处。好处是异常信息不会丢你能通过getCause()找到原始异常。坏处是如果你没有显式处理异常异常会沿调用链一路传递到最终的join()然后以CompletionException的形式抛出来调用方如果只catch了Exception可能容易漏。解决方法是在每个可能出错的节点上尽量显式处理。比如token获取失败就用exceptionally兜底单个消息发送失败在内部try-catch处理。这样异常就在最近的地方被消化掉不会污染后面的编排链。5.2 偶发卡住的排查思路有一次线上任务运行时间从正常的30秒突然变成5分钟我排查了很久。最后发现是某个下游依赖出现了网络抖动导致部分消息发送请求迟迟不返回。而代码里发送客户端的HTTP超时时间设置得过大默认甚至可能是无限等待。这个问题有两个层次的解法。第一层在HTTP客户端层设置合理的连接超时和读取超时比如连接超时3秒读取超时5秒不让单个请求无限阻塞下去。第二层在CompletableFuture层面增加整体超时控制使用orTimeout方法summaryFuture summaryFuture.orTimeout(60, TimeUnit.SECONDS);orTimeout在指定时间内没有完成会让future以TimeoutException异常状态完成配合exceptionally可以返回一个超时标记的汇总结果。这个方法在Java 9开始提供如果你还在用Java 8需要自己用ScheduledExecutorService配合实现或者升级版本。我当时是直接用Java 11环境所以可以放心用。5.3 并发修改共享状态的坑SendContext里的successCount和failCount我用了AtomicIntegerfailedMessages用了ConcurrentLinkedQueue。为什么不用普通类型因为多个发送子任务是在不同线程里同时执行的会同时往这些字段里写数据。普通ArrayList在多线程并发add会导致数据错乱甚至数组越界普通int自增在并发下会丢更新。我建议凡是会被多个CompletableFuture子任务共享修改的数据结构一律用并发安全的版本。不要图省事用普通集合最后排数据问题的成本远比改一行代码高。5.4 幂等与去重设计企业微信发送消息偶尔会因为网络重传导致同一条消息被发送两次。这在业务上可能造成用户收到重复通知体验很差。我的方案是给每条消息生成一个唯一消息ID发送前检查发送记录表里是否已有该ID的成功记录如果已存在则跳过。这个检查同样放在发送子任务内部用并发安全的ConcurrentHashMap做已发送ID的缓存。这个设计比较简单但能挡住大部分重复发送的场景。5.5 排查清单速查表症状可能原因解决思路任务执行时间异常长单次HTTP请求超时未拦截检查HTTP客户端超时参数整体加orTimeout出现大量发送失败并发过高触发频控调小线程池最大线程数检查分批策略内存占用飙升队列过大或消息体过大控制队列容量压缩消息内容部分批次没有执行线程池拒绝策略丢弃任务改用CallerRunsPolicy结果统计不准共享状态未用并发安全类换AtomicInteger、ConcurrentLinkedQueuejoin抛出CompletionException链路中某节点异常未被捕获检查每个节点的exceptionally兜底这个表是我在多次故障排查中总结出来的每次遇到任务表现异常先从这几个方向排查大部分问题都能快速定位。6. 性能收益与使用感受改造完成上线后我统计过一组对比数据。同样的5000条企业微信文本消息旧的串行方案平均耗时约12分钟裸线程池并发方案约3分钟但伴随偶发频控CompletableFuture编排方案稳定在40秒到60秒之间而且没有触发过频控错误。对于1000条以内的常规推送基本20秒内完成。这个提升有两部分来源一部分是并发带来的吞吐量提升这个多线程本来就能做到另一部分是编排带来的稳定性收益这体现在失败重试更精准、频控规避更主动、异常处理更连贯。后者虽然不直接体现在耗时时长上但体现在运维成本和业务体验上。另外还有一个不容易量化的收益代码可维护性。改造后整个发送流程的代码量没有增加多少但逻辑清晰度肉眼可见地提升了。之前串行方案里散落在各处的失败处理、重试逻辑现在都集中在编排链路的各个节点上新同事接手时看一遍代码就能理解全流程。7. 日常习惯层面的几点建议最后说几个我在实践中沉淀下来的习惯。不一定每个都适合所有团队但应该能给准备上手CompletableFuture的开发者一些启发。第一不要为了用而用。如果你的业务场景只有一个异步任务不需要编排直接ThreadPoolExecutor提交就够了。CompletableFuture的真正价值出现在任务之间有依赖关系、有并发分叉、有汇合点的时候。批量消息发送正是这种场景的典型代表。第二设计异步链路之前先画一张任务依赖图。哪怕画在草稿纸上也行标清楚哪些任务可以并行、哪些必须等前一步完成、哪里有分支和合并。这张图定了写代码就是照图施工。跳过这一步直接写代码后期会反复调整编排逻辑很痛苦。第三理解每个组合方法的语义差异。thenApply、thenCompose、thenAccept、thenRun、whenComplete、exceptionally这几个方法使用场景完全不同。花时间吃透它们的区别比背一百个示例代码更有用。第四测试时一定要模拟异常场景。我见过很多异步代码在正常路径下跑得很好一旦某个环节抛异常整个链路就乱了。在设计阶段就给每条链路标注异常出口测试阶段人为制造异常验证兜底逻辑是否真的有效。CompletableFuture是一个值得深入掌握的类。它把异步编排从“手工管理线程和状态”提升到“描述任务依赖图”的层次代码质量和运行效率都能得到明显改善。如果你也在处理类似的批量任务场景希望这篇文章能帮你少踩几个坑。