Java实现CompletionService并发编排消费任务

这是很多人都遇到过的场景：RocketMQ 一次性拉取了 128 条消息，可消费端是一条一条串行处理的。每条消息耗 50 毫秒，一轮下来就是 6.4 秒。听着就让人头疼，对吧？

那能不能直接用批量消费来加速呢？

RocketMQ 的 setConsumeMessageBatchMaxSize(128) 确实允许你一次性拉 128 条消息，但如果还是逐条同步处理，那批量拉取的意义基本就只剩下“减少网络往返”了。处理速度的瓶颈，一条都没解开。

直觉上的解法很简单：交给线程池并发执行不就完了？但这里面藏着一个很容易被忽略的问题。

如果异步线程执行失败了，Broker 是毫不知情的。主线程已经返回了 CONSUME_SUCCESS，Broker 提交了偏移量，那条失败的消息就无声无息地丢了。更常见的情况是，主线程根本搞不清楚哪些子线程成功了、哪些失败了，只能盲目地返回成功或重试。

这里缺的是一个办法：让主线程能够感知每一个子线程的执行结果。全部成功才返回 CONSUME_SUCCESS，有一条失败就返回 RECONSUME_LATER，让 RocketMQ 整体重发。

问题的核心不是“能不能并发”，而是“并发了之后能不能控得住”。

Ja va 并发包里有一个接口叫 CompletionService，在 ja va.util.concurrent 包下，专门干这件事：批量提交异步任务，然后按完成顺序逐个取结果。

它的唯一实现类是 ExecutorCompletionService，用法就三步。

1. 提交所有任务到线程池。
2. 循环取结果，发现失败立即标记。
3. 全部成功返回 CONSUME_SUCCESS，否则返回 RECONSUME_LATER。

代码逻辑是这样的。

@Override
public void prepareStart(DefaultMQPushConsumer consumer) {
    consumer.setPullInterval(1000);
    consumer.setConsumeMessageBatchMaxSize(128);
    consumer.setPullBatchSize(64);
    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
        log.info("NewBuyBatchMsgListener receive message size: {}", msgs.size());

        CompletionService completionService = new ExecutorCompletionService<>(executor);
        List> futures = new ArrayList<>();

        // 1. 提交所有任务
        msgs.forEach(messageExt -> {
            Callable task = () -> {
                try {
                    OrderCreateRequest orderCreateRequest = JSON.parseObject(JSON.parseObject(messageExt.getBody()).getString("body"), OrderCreateRequest.class);
                    return doNewBuyExecute(orderCreateRequest);
                } catch (Exception e) {
                    log.error("Task failed", e);
                    return false; // 标记失败
                }
            };
            futures.add(completionService.submit(task));
        });

        // 2. 检查结果
        boolean allSuccess = true;
        try {
            for (int i = 0; i < msgs.size(); i++) {
                Future future = completionService.take();
                if (!future.get()) { // 3. 发现一个失败立即终止
                    allSuccess = false;
                    break;
                }
            }
        } catch (Exception e) {
            allSuccess = false;
        }

        // 3. 根据结果返回消费状态
        return allSuccess ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS
                : ConsumeConcurrentlyStatus.RECONSUME_LATER;
    });
}

128 条消息并发处理，总耗时从 6.4 秒直接降到最慢那条消息的耗时，通常几十毫秒就搞定了。

那么，CompletionService 底层是怎么做到“先完成先取”的？

ExecutorCompletionService 的源码非常精炼，核心就三个成员变量。

private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue> completionQueue;

executor 是你传入的线程池，completionQueue 是一个 LinkedBlockingQueue，用来存放已完成任务的 Future 对象。

关键在于 submit 方法里。当你调用 completionService.submit(task) 时，它并没有直接把 task 丢给线程池，而是先包装了一层。

private class QueueingFuture extends FutureTask {
    QueueingFuture(RunnableFuture task) {
        super(task, null);
        this.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future task;
}

QueueingFuture 继承自 FutureTask，重写了 done() 方法。done() 是 FutureTask 提供的钩子，任务无论正常完成还是异常终止，都会回调这个方法。

所以整个流程是这样的：你 submit 一个任务，它被包装成 QueueingFuture 交给线程池执行。任务跑完的那一刻，done() 触发，把对应的 Future 塞进 completionQueue。你调 take()，就是从 completionQueue 阻塞地取一个出来。

谁先完成，谁的 Future 先入队，你就先取到谁的结果。提交顺序和完成顺序被解耦了。

这个设计非常漂亮。它没有用任何锁排序、没有用优先队列、没有用回调链，就是最朴素的“生产者往队列里放，消费者从队列里取”。BlockingQueue 天然线程安全，生产消费解耦，简单到几乎不可能出错。

Ja va 8 引入了 CompletableFuture，同样是处理异步任务的利器。它和 CompletionService 解决的问题有重叠，但设计哲学完全不同。

一句话总结：CompletionService 是“批量并发，按完成顺序收结果”；CompletableFuture 是“异步编排，按依赖关系串流程”。

回到 RocketMQ 批量消费的场景，128 条消息之间没有任何依赖关系，我们只关心“全部成功还是有一个失败”。这正是 CompletionService 的主场。

如果你用 CompletableFuture 来写，也能做，但需要自己维护一个 CompletableFuture.allOf() 来等全部完成，然后再遍历检查结果。代码更啰嗦，而且 allOf 会等所有任务都完成才能继续——哪怕第 2 条消息就失败了，你也要等剩下 126 条跑完才能返回。CompletionService 的 take() 则是逐个检查，发现失败立即 break，省下了不必要的等待。

当然，如果你的场景是“查商品信息，再根据商品查库存和价格，最后组装结果”，任务之间有明确的先后依赖，那 CompletableFuture 的链式编排就比 CompletionService 的队列取值优雅得多。

工具没有好坏，只有合不合适。

串行消费慢，直觉反应是加并发。但加了并发之后，如果主线程无法感知子线程的成败，那并发就不是加速，是埋雷。消息丢了都不知道。

CompletionService 解决的不是“怎么并发”的问题，而是“并发了怎么收场”的问题。它用最朴素的 BlockingQueue 机制，让主线程能按完成顺序逐个检查结果，发现异常立即止损。

并发编程的难点，从来不是“怎么让任务跑起来”，而是“跑起来之后怎么确保结果可控”。CompletionService 给了一个很干净的答案。