做企业微信服务端开发这些年，最常被运营同事吐槽的就是群发消息太慢。三千人的通讯录按部门一批批发下去，接口一个接一个地等，高峰期最长要跑到两三分钟，用户早就开始刷屏问"消息是不是丢了"。后来我把整个发送链路重新拆了一遍，用 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 List<String> userIds; private final String content; private final String tag; // 用于标记部门或批次来源 public WxSendTask(List<String> 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 List<SendResult> batchSend(List<WxSendTask> tasks, long timeoutMillis) { List<CompletableFuture<SendResult>> futures = new ArrayList<>(); for (WxSendTask task : tasks) { CompletableFuture<SendResult> future = CompletableFuture .supplyAsync(() -> wxApiClient.send(task), sendExecutor) .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) .exceptionally(ex -> SendResult.failed(task, ex.getMessage())); futures.add(future); } CompletableFuture<Void> 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 的异常都要有明确处理路径，最怕的就是异常被吞掉却毫无感知。做到这三点，批量发送这类任务基本就能跑得又快又稳了。
这套方案后续还可以继续扩展，比如结合动态线程池参数调优、将失败补偿表接入定时任务实现最终一致性，或者把任务拆分配置化。不过那是下一步的事了，先把眼前的消息发稳，比什么都重要。
