做企业微信服务端开发这些年最常被运营同事吐槽的就是群发消息太慢。三千人的通讯录按部门一批批发下去接口一个接一个地等高峰期最长要跑到两三分钟用户早就开始刷屏问消息是不是丢了。后来我把整个发送链路重新拆了一遍用 Java 的 CompletableFuture 做异步编排同样的量级压缩到十几秒系统负载反而降了。这篇文章就把当时的改造思路、关键代码和踩过的坑完整分享出来适合正在做企业微信应用开发、或者想优化批量任务耗时的后端同学参考。先说清楚这套方案的适用范围消息总量在几百到几万条之间对实时性要求没那么苛刻但又不能慢到让用户感知明显调用的是外部 HTTP 接口天然适合并发并且你希望代码还能保持可读、可维护而不是上来就维护一堆手写线程池加 Future 的轮询逻辑。如果你正好卡在这个场景里CompletableFuture 基本就是最顺手的工具。1. 为什么需要异步编排先看一下原来的串行痛点1.1 一条消息从发出到送达经历了什么大多数企业内部的企业微信应用发消息走的是应用消息推送接口。调用一次传入 touser、toparty、agentid 和消息体接口返回 errcode 告诉你有没有推给微信客户端。问题就出在批量这两个字上通讯录可能几千人企业微信接口单次调用有接收人数量上限超过就得拆批不同部门还要发不一样的内容又得按部门分组下发。原来的实现很朴素一个 for 循环把每个批次按顺序发给企业微信接口等它返回后再发下一批。一次请求 1 秒左右一百批就是一百秒放到用户端就是消息半天没收到。而且这个等待时间全部占用了处理线程本来可以做别的事情也被夹在中间动弹不得。更尴尬的是如果中间某几批失败需要重试整体耗时还要再往上加。这种串行模型的问题不是单点慢而是把大量独立的小任务强行排成了一队。部门 A 和部门 B 的消息之间没有任何依赖关系完全可以同时发批次 1 和批次 2 之间也不需要知道彼此的结果却非得等对方走完。这种场景几乎就是为异步编排量身定做的。1.2 CompletableFuture 解决的核心问题CompletableFuture 最直接的价值是把发起调用和等待结果解耦。发起调用时任务被丢给线程池立刻执行主线程不用傻等后续需要拿结果做处理时再通过回调函数把逻辑接上去。这样一批消息的 HTTP 请求可以在多个线程上同时跑总量不变但墙钟时间被大幅压缩。更关键的是它把编排这件事做成了标准 API。批量发送里有几个固定动作拆分批次、并发发送、等待所有批次完成、汇总结果、处理失败项。用 CompletableFuture 的 allOf 可以等所有任务结束用 thenApply / thenCompose 可以串联依赖步骤用 exceptionally 可以给失败分支兜底。整个流程可以用声明式的方式描述出来而不是在代码里写一堆 get() 阻塞等待和状态判断。我个人更看重它的一点异常信息不会丢。以前用线程池加 Future要么 get() 时抛 ExecutionException要么线程里 try-catch 把异常吞了日志都看不到。CompletableFuture 把每个任务的结果和异常都封装在 future 里聚合时统一处理这对排查线上问题帮助很大。1.3 你不需要它的几个场景也不是所有批量发送都非得上 CompletableFuture。如果消息总量很小比如一次就几十个人串行调几次接口也就一两秒没必要引入异步复杂度。如果你的场景是真正意义上的高吞吐、需要削峰填谷那正确的选择是消息队列而不是在业务代码里堆线程。另外还有一种情况要避开如果下游接口已经做了严格的并发控制比如只允许两个并发那你用 CompletableFuture 拉满并发反而会触发限频。异步编排的前提是下游扛得住压力或者你自己手里有可靠的限流手段。这一点在后面实操部分会专门讲。2. 动手前必须搞懂的 API 与线程模型2.1 三个最常用的 API 组合CompletableFuture 的 API 数量很多乍看容易劝退但实际批量发送场景里高频使用的就几个。supplyAsync 是入口负责把一个带返回值的任务丢到线程池执行。它内部会包装成 CompletableFuture你在 lambda 里写具体的 HTTP 调用逻辑返回值就是接口的响应体。thenApply 用于串行转换等上一个 future 完成后把结果拿过来处理再产生一个新的 future。比如发送返回后解析 errcode、提取需要重试的批次就很适合放在 thenApply 里。allOf 是批量编排的主角。它接收一堆 CompletableFuture返回一个新的 CompletableFuture当所有传入的 future 都完成时才完成。注意它返回的是 CompletableFuture 本身不携带每个子任务的结果所以通常配合一个共享的结果容器使用。这三个 API 对应了批量发送的三个必做动作发起并发调用、逐个处理响应、等全部结束后收尾。把它们组合起来已经能覆盖绝大部分需求。2.2 回调到底跑在哪个线程这是新手最容易踩的坑。supplyAsync 传入的任务一定在指定的线程池里执行但 thenApply、whenComplete 这些回调的执行线程并不固定。如果没有指定第二个 Executor 参数回调会在前面那个任务完成的线程上继续执行如果前面的任务已经完成回调甚至可能直接在调用方线程里跑。这个行为带来的问题很隐蔽。假设你在 allOf 后面接了一个 thenApplyAsync 做结果统计但忘了传线程池那么统计逻辑可能跑在最后一次 HTTP 调用的线程上。如果此时线程池的线程都在干活统计代码会挤占执行位反而拖慢后续任务的调度。我的习惯是所有可能阻塞的后续操作都显式传入线程池例如 thenApplyAsync(x - {...}, sendExecutor)。前面提到的 thenApplyAsync 也一样。宁可多打几个字也不要把线程模型交给默认规则去猜。默认的 ForkJoinPool.commonPool 在普通业务应用里千万别用它被其他框架共用阻塞一个任务可能影响全局。2.3 线程池参数IO 密集型任务的默认配置企业微信消息发送是典型的 IO 密集操作线程大部分时间在等 HTTP 响应。这种任务线程数可以适当放大参考公式大约是 CPU 核数乘以 2 到 4考虑到每线程还有内存开销也不要盲目拉大。当时线上机器是 8 核我配置的核心线程数 16最大线程数 32队列长度 200。核心线程数保证平时批量任务不需要频繁创建线程最大线程数要留出余量避免瞬时大任务进来后线程扩张过猛队列长度不能太长否则任务积压会让消息延迟变得不可控。线程池的拒绝策略要自定义。默认的 AbortPolicy 会直接抛异常如果发生在 supplyAsync 内部任务会以异常完成最后通过 exceptionally 被捕获。可以考虑 CallerRunsPolicy让提交任务的线程自己执行这样至少不会丢消息只是速度略降。不过 CallerRunsPolicy 也有副作用它会让主流程线程被阻塞实际用的时候要看整体链路能不能接受我是配合信号量限流一起用的。线程池要给个有意义的名称。好记的线程工厂能救命比如 wx-send-pool-thread-这样出现线程卡死时在线程 dump 里一眼就能识别故障来源。2.4 超时控制不会等待的编排才有意义异步最大的风险是任务永远不结束。HTTP 调用方如果没设置超时future 就一直是 pending 状态allOf 也跟着永远等下去消息积压最终拖垮整个线程池。所以每层都要有超时。底层 HTTP 客户端设置连接超时和读取超时例如连接 1 秒、读取 3 秒上层编排再补一道保险用 orTimeout 给未来加时间限制到点还没完成就让它异常结束。Java 9 的 completeOnTimeout 也可以允许到点后返回一个默认值但要注意这个默认值必须和正常结果类型兼容不适合做复杂结果时慎用。另一个容易忽略的点get() 方法的超时。如果代码里最终要同步拿结果建议用 future.get(5, TimeUnit.SECONDS)不要让阻塞没有上限。即便整体是异步编排收尾汇总那一步通常也需要一个明确的等待上限。3. 批量消息发送的完整异步编排实现3.1 整体流程拆解拆分、并发、聚合、补偿实现前先画出流程从数据库或接口拿到全量员工和部门信息按部门分组再按单批上限拆分成多个发送任务每个任务对应一次企业微信接口调用并发执行所有任务完成后统一汇总发送成功、失败、需要重试的批次失败的批次进入重试流程重试仍失败的写入补偿表由定时任务兜底。这个流程的关键点在于拆分后任务之间无依赖这是并发安全的前提。所有任务只读取入参写结果时各自写独立的 Future最后由汇总逻辑合并不存在共享可变状态。批量上限要根据企业微信接口的约束来定。单次接收人太少会放大请求次数太多则可能触发接口限制。实测按 100 人一批比较稳三千人就是三十批三十二线程的池子基本可以同时跑完。3.2 分批并发发送的核心代码下面是一段简化后的核心代码保留了主要逻辑结构实际项目中请根据自身封装调整。先定义发送任务类public class WxSendTask { private final ListString userIds; private final String content; private final String tag; // 用于标记部门或批次来源 public WxSendTask(ListString userIds, String content, String tag) { this.userIds userIds; this.content content; this.tag tag; } // getter ... }发送任务实现通过企业微信接口发送消息并返回结果public class WxMessageSender { private final ExecutorService sendExecutor; private final WxApiClient wxApiClient; public WxMessageSender(ExecutorService sendExecutor, WxApiClient wxApiClient) { this.sendExecutor sendExecutor; this.wxApiClient wxApiClient; } public ListSendResult batchSend(ListWxSendTask tasks, long timeoutMillis) { ListCompletableFutureSendResult futures new ArrayList(); for (WxSendTask task : tasks) { CompletableFutureSendResult future CompletableFuture .supplyAsync(() - wxApiClient.send(task), sendExecutor) .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) .exceptionally(ex - SendResult.failed(task, ex.getMessage())); futures.add(future); } CompletableFutureVoid allDone CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); // 等所有任务完成但最多再等一段时间防止 orTimeout 失效等极端情况 allDone.get(timeoutMillis 2000, TimeUnit.MILLISECONDS); return futures.stream() .map(CompletableFuture::join) .collect(Collectors.toList()); } }有几个细节要说明。supplyAsync 的返回类型是 CompletableFuture 或者超时后包装的异常结果都是同一类型所以 exceptionally 里返回 failed 结果不会类型错乱。orTimeout 在 Java 9 以上可用如果项目还在 Java 8就得用 get(timeout) 或者自己写延时任务来模拟超时。最后一个 join 不会真的阻塞很久因为前面已经等了 allDone正常情况下子任务都已完成。之所以不用 get 而用 join是为了避免处理 InterruptedException 和 ExecutionException 的样板代码因为异常已经被 exceptionally 捕获并转换掉了。3.3 token 刷新与限流的并发控制调用企业微信接口需要 access_token这个 token 有有效期通常两小时。批量发送的并发场景下如果 token 刚好过期会有大量请求同时拿到过期 token 然后失败如果代码里没有做好并发控制还会出现多个线程同时去刷新 token 的情况刷新接口被调用多次白白浪费资源。我的做法是用一个 AtomicReference 保存当前 token加一个可重入锁保护刷新动作。发送时先从 AtomicReference 读取 token如果调用返回 token 无效的错误码则尝试获取锁并刷新 token。获取不到锁的线程说明已经有其他线程在刷新了等锁释放后重新读取一次即可。另外就是限流。企业微信接口有频率限制并发太高会被 45009 之类的频控错误码打回。即使线程池大小固定业务高峰期任务也可能全部积压后突然涌出造成瞬时流量波峰。我在发送任务外层包了信号量 Semaphore限制同时进行中的 HTTP 调用数量信号量不足时任务会在 supplyAsync 内部短暂等待起到削峰作用。private final Semaphore sendSemaphore new Semaphore(20); public SendResult sendWithLimit(WxSendTask task) { boolean acquired false; try { sendSemaphore.acquire(); acquired true; return wxApiClient.send(task); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return SendResult.failed(task, interrupted); } finally { if (acquired) { sendSemaphore.release(); } } }信号量的数量和线程池最大线程数需要配合起来。我的线程池最大 32信号量 20意味着最多同时有 20 个请求在飞剩下的线程可能在等待信号量但不至于把 HTTP 连接池打爆。这个值取决于企业微信接口的具体频控上限需要实际压测后调整。3.4 失败重试与结果统计批量发送不可能每次都全部成功网络抖动、token 过期、接口频控都会导致部分批次失败。重试策略我采用了简单指数退避第一次失败后等 500 毫秒重试再失败等 1 秒最多重试三次。这个策略不需要引入额外依赖执行重试时直接用 ScheduledExecutorService 或者 Thread.sleep 都可以。考虑到重试次数少我在代码里使用的是 Thread.sleep配合异步线程池不至于阻塞主流程。重试逻辑放在 exceptionally 里不好写因为异常处理分支拿不到原始任务的上下文。我更推荐在发送逻辑内部封装一层带重试的调用方法重试全部失败后才抛出异常由外层的 exceptionally 统一转换成失败结果。结果统计要落到指标上不能只打日志。我用了简单的计数器总数、成功数、失败数、平均耗时、p99 耗时。这些数据上报到监控系统后能清楚看到每次群发的质量。特别是失败率突然升高时能第一时间发现是企业微信接口频控还是 token 配置出了问题。4. 常见问题与排查技巧实录4.1 现象一线程池被打满HTTP 调用全部超时上线初期我把线程池最大线程数调得很高以为并发越大越快。结果某次全员通知时线程池迅速扩张所有线程同时发起 HTTP 调用下游连接池被打爆大量请求超时接着触发重试重试又占用新线程形成恶性循环最终消息延迟比改造前还严重。排查时从线程 dump 看到大量线程停在 HTTP 调用上配合监控里的 TCP 连接数确认是连接池耗尽。这个问题的根因不是 CompletableFuture 本身而是没有给并发设置边界。后来加了信号量限流并把线程池最大线程数调低同时给 HTTP 客户端设置了严格的最大连接数问题才彻底解决。核心经验是异步编排只是把控制权交给你并发边界的责任也在你。4.2 现象二allOf 之后回调迟迟不执行有一次发完消息后统计结果迟迟没有输出日志里也没有异常。排查后发现是因为某个发送任务没有设置超时底层 HTTP 调用一直挂着导致对应的 future 永远不完成allOf 也就永远不完成。处理办法就是前面提到的双保险HTTP 层设置读取超时编排层加 orTimeout。另外还要注意 allOf 返回的 future 虽然叫 allOf但它本身也可能异常完成建议给它也加上 exception 处理避免因为个别任务异常导致整个收尾逻辑不跑。如果遇到回调不执行但所有子任务看起来都完成了还有一个可能回调方法本身抛了异常异常被 CompletableFuture 捕获后导致后续依赖该回调的流程中断。这种问题打日志最有效在回调里第一行就记录参数和执行结果能快速定位。4.3 现象三异常静默吞掉线上看不到失败CompletableFuture 的异步任务里如果抛出异常且没有对应的 exceptionally 或 whenComplete 处理异常不会直接打到日志里只会让 future 以异常状态完成。如果后续代码忽略了 future 的状态或者直接 join 没有处理异常失败就好像被吞掉了一样。踩过一次这个坑之后我在所有入口 supplyAsync 的 lambda 里都包了一层 try-catch异常转换成自定义的 SendResult.Failed同时保留异常堆栈打印。这样不管外层编排有没有正确处理异步状态日志里至少能看到失败原因。后续如果引入调用链追踪可以在异常转换时把 traceId 放进去排查效率会更高。4.4 现象四明明控制了并发还是被企业微信限频信号量和线程池都设了但还是收到频控错误码。后来发现限频不是按 QPS 算的而是按分钟累计调用量算的。瞬时并发控制住了但如果一个批量任务连续跑几分钟累计调用量很快就撞上上限。解决思路是把批量任务再分片每片之间加一个短暂间隔或者把总任务量拆成多轮每轮之间根据已消耗的配额动态调整速度。实现上不难但需要你对自己企业应用的频控配额有明确认知最好去企业微信后台确认或者在测试环境压出实际阈值。这个数值不同应用差别很大不能照搬别人的配置。4.5 快速排查方法把编排链路可视化CompletableFuture 的编排代码在出现问题时排查难度比同步代码高因为调用链被拆散了。我的办法是给每个批次打上唯一批次号所有日志统一输出批次号、任务 ID、线程池线程名。出现问题时直接按批次号把日志拼起来就能还原整个编排过程。如果日志还不够就在代码里给每个 future 加 tag例如把业务含义存进任务对象异常时打印 tag 而不是打印一堆对象地址。另外关键节点可以埋点统计进入线程池前的等待时长、HTTP 调用时长、重试次数、最终失败数。这几项数据能帮你快速判断瓶颈在编排层还是在下游接口。线程池的拒绝次数、活跃线程数、队列深度也要盯。这些指标结合 allOf 完成时的任务数量可以看出一轮批量任务是否都顺利跑完还是有一部分压根没进线程池就被拒掉了。监控不一定需要多复杂的系统我一个本地开发机上用 JDK 自带的 jstat 加日志统计也够用。写在最后一点使用心得改造完这套异步编排后我最直观的感受是代码结构反而比原来更清晰了。串行 for 循环里塞满业务逻辑的写法改成任务拆分加异步编排后每一步做什么都很明确后来接手这块代码的同事也说好维护。但我必须提醒一点CompletableFuture 不是银弹它要求你对线程模型、超时、并发边界都有清晰的认识否则出问题的时候比同步代码难查得多。如果让我给出几个最重要的建议第一所有异步入口都要有超时兜底第二线程池和信号量的参数一定要压测别拍脑袋第三每个 future 的异常都要有明确处理路径最怕的就是异常被吞掉却毫无感知。做到这三点批量发送这类任务基本就能跑得又快又稳了。这套方案后续还可以继续扩展比如结合动态线程池参数调优、将失败补偿表接入定时任务实现最终一致性或者把任务拆分配置化。不过那是下一步的事了先把眼前的消息发稳比什么都重要。