《Java 消息中间件·可靠消息与消费治理》12.1 消费者多线程模型

一个消费者跑满一核，lag 还是不降

libraryd 的借阅投影消费组订阅 library.loan.created，四个分区。进程只有一个消费者，把 max.poll.records 从 500 调到 5000 之后，top 里这个 JVM 稳定在一个核附近，lag 却以同样的斜率往上走。下游的投影更新是纯内存计算，不慢；慢的是「一个线程既要拉取、又要反序列化、又要处理、又要提交」。

直觉的解法是加线程，于是有了两种常见写法：

• 开 N 个线程，每个线程 new 一个 KafkaConsumer，各自负责一部分分区；
• 只留一个 KafkaConsumer 拉取，把记录丢给线程池处理。

两种写法都能把 CPU 用起来，但它们在顺序、提交和线程数上的代价完全不同。本篇把两种写法各写一份可运行实现，再加上一个折中的「按分区亲和路由」版本，用同一批消息跑一遍，把吞吐、线程数和乱序次数摆出来。

先确认约束：消费者不是线程安全的

KafkaConsumer 的类注释写得很直接：

/**
 * The consumer is not thread-safe. See <a href="#multithreaded">Multi-threaded Processing</a> for more details.
 */

ClassicKafkaConsumer 用一个「轻锁」实现这个约束：同一条线程可以重入（refcount 计数），换一条线程进来就抛 ConcurrentModificationException。

private void acquire() {
    final Thread thread = Thread.currentThread();
    final long threadId = thread.getId();
    if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
        throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access. " +
                "currentThread(name: " + thread.getName() + ", id: " + threadId + ")" +
                " otherThread(id: " + currentThread.get() + ")"
        );
    refcount.incrementAndGet();
}

所以「多个线程共享一个 consumer」这条路从一开始就不成立。能共享的是「拉到的记录」，不是 KafkaConsumer 本身。

另外，4.3.0 默认仍然走 classic 协议（ConsumerConfig.DEFAULT_GROUP_PROTOCOL = classic），KafkaConsumer 门面背后是 ClassicKafkaConsumer。本篇的结论都基于默认配置；改成 group.protocol=consumer 走 KIP-848 的实现后，线程模型不变，协调细节会变。

三种模型的分工

三者的共同点是「一个消费者只被一条线程使用」。区别在于记录被谁处理、顺序在哪里断掉。

实验装置

环境：Temurin 25.0.5+8、kafka-clients 4.3.0、apache/kafka:4.3.0 单节点 KRaft，宿主机 4 vCPU。以下数字均为单次运行，随调度波动，不作基准。

实验数据：

• 主题 4 个分区，每分区 40000 条，共 160000 条；
• 每个分区固定一个 key（book-0 到 book-3），value 形如 book-2,<seq>，seq 从 0 递增；
• 每条记录做同样的模拟处理：20000 次 xorshift 位运算，约几十微秒；
• 乱序判定：维护每个 key 上一次见到的 seq，新来的 seq 更小就计一次逆序。

三种模式用同一个 main，切换实现：

JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
java -Xmx1g -cp "out:$(cat cp.txt)" ConsumerThreadingLab perthread 4
java -Xmx1g -cp "out:$(cat cp.txt)" ConsumerThreadingLab sharedpool 4
java -Xmx1g -cp "out:$(cat cp.txt)" ConsumerThreadingLab affinity 4

每种模式跑三轮。线程数由一条后台采样线程每 20 毫秒统计一次线程名得到，避免把 JVM 启动阶段的峰值算进来。

模型 A：每线程一个消费者

每个线程创建自己的 KafkaConsumer，用 assign 锁定一个分区，自己 poll、自己处理、自己提交。

for (TopicPartition tp : partitions(topic)) {
    Thread t = new Thread(() -> {
        Properties p = baseProps(group + "-" + tp.partition());
        try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
            c.assign(List.of(tp));
            c.seekToBeginning(List.of(tp));
            long end = endOffsets.get(tp);
            long committedAt = 0;
            while (true) {
                ConsumerRecords<String, String> rs = c.poll(Duration.ofMillis(200));
                for (ConsumerRecord<String, String> r : rs) {
                    handle(r);            // 业务处理 + 顺序检查
                }
                if (c.position(tp) - committedAt >= 2000) {
                    c.commitSync();
                    committedAt = c.position(tp);
                }
                if (c.position(tp) >= end) break;
            }
            c.commitSync();
        }
    }, "consumer-" + tp.partition());
    t.start();
}

实测（三轮）：

RESULT mode=perthread records=160000 processed=160000
elapsedMs=1853 throughput=86304 rec/s
threads maxConsumers=4 maxWorkers=0
ordering inversions=0

三轮耗时分别是 1853、1913、1876 毫秒，吞吐 86k、84k、85k 条每秒，逆序始终为 0。

这个结果是可预期的：每个分区只有一条线程在碰它，处理顺序就是拉取顺序，拉取顺序就是分区内的位移顺序。提交也在同一条线程上完成，没有任何跨线程交接。

代价也清楚：四个消费者实例，四份 fetch 缓冲，四条到协调器的连接。分区数就是并行度的上限——想再快只能加分区，加线程没有用。

模型 B：一个拉取线程 + 共享工作池

只留一个消费者，poll 到的记录逐条提交给固定大小的线程池；拉取线程在每轮 poll 之后只提交「连续完成」的位移。

while (true) {
    ConsumerRecords<String, String> rs = c.poll(Duration.ofMillis(200));
    for (ConsumerRecord<String, String> r : rs) {
        Future<?> f = pool.submit(() -> handle(r));
        pending.get(new TopicPartition(r.topic(), r.partition()))
                .addLast(Map.entry(r.offset() + 1, f));
    }
    // 只推进“队首已经完成”的位移，提交点不越过未完成记录
    for (TopicPartition tp : tps) {
        Deque<Map.Entry<Long, Future<?>>> q = pending.get(tp);
        while (!q.isEmpty() && q.peekFirst().getValue().isDone()) {
            commitMark.put(tp, q.pollFirst().getKey());
        }
    }
    if (allCaughtUp(c, tps, ends) && allDone(pending)) break;
}
c.commitSync(offsets(commitMark));

实测（三轮）：

RESULT mode=sharedpool records=160000 processed=160000
elapsedMs=2294 throughput=69730 rec/s
threads maxConsumers=0 maxWorkers=4
ordering inversions=6329 first=key=book-2 saw=2 after=5

三轮的逆序次数分别是 6329、3276、5205，样例是「先看到 book-2 的 seq=5，又看到 seq=2」。这不是偶发：同一个 key 的相邻两条记录被提交给池里不同的 worker，谁先跑完谁先记账，顺序就断了。

吞吐比模型 A 低约 19%（69k 对 85k）。原因不是 worker 少，而是多了一层交接：submit 创建任务、Future 记账、跨线程队列的缓存行来回搬。单看「把 CPU 用满」这件事，共享池并不比每线程一个消费者更划算。

还有一个容易被忽略的点：上面这段代码里，提交的是「连续完成的位移」。如果图省事，poll 完直接 commitSync(最后一条的 offset + 1)，提交点就会跑到处理前面——这正是 12.2 要复现的丢消息场景。

模型 C：按分区亲和路由

既然乱序来自「同一 key 落到不同 worker」，那就把路由固定下来：每个分区一个单线程 lane，记录按 tp.partition() 投递。与模型 B 相比只改一行：

// 模型 B：共享池，任意 worker 都可能拿到这条记录
Future<?> f = pool.submit(() -> handle(r));

// 模型 C：按分区固定 lane，同一分区的记录永远排在同一条线程上
Future<?> f = lanes.get(tp.partition())
        .submit(() -> handle(r, laneLast.get(tp.partition())));

实测（三轮）：

RESULT mode=affinity records=160000 processed=160000
elapsedMs=2331 throughput=68615 rec/s
threads maxConsumers=0 maxWorkers=4
ordering inversions=0

吞吐 68.6k、69.8k、69.9k，与共享池几乎一样；逆序为 0。说明「保序」本身不额外花钱，真正的成本在跨线程交接，而这两种模型都有。

三组数字放在一张表里

三个模型用的处理函数、消息总量、分区数完全一致，差异只来自线程模型。可以读出三件事：

1. 每线程一个消费者最快。省掉了任务提交和跨线程交接，单核效率最高。
2. 共享工作池不天然更快。它的收益在下游是阻塞 IO 时才会出现——worker 被 IO 挂起时，poll 线程还能继续拉取。本实验的处理是纯 CPU，池化只有成本没有收益。
3. 保序不贵，乱序很贵。C 与 B 的吞吐只差 1%，但一个逆序为 0，一个有几千次。顺序是路由决定的，不是线程数决定的。

顺序是怎么被保住的

Kafka 只保证分区内有序。消费者一旦把记录交给多条线程，「分区内有序」这个前提就失效了。要在多线程下保住顺序，需要同时满足三个条件：

1. 同一个 key 永远路由到同一条线程。按 key 哈希取模，或者干脆按分区路由（key 已经决定了分区）。
2. 这条线程串行处理。lane 内部不能再开池，否则又回到模型 B。
3. 提交点不越过未完成记录。提交逻辑必须看「连续完成」的位置，而不是「已经 poll 到」的位置。

前两条决定处理顺序，第三条决定崩溃恢复后的重放范围。只做前两条，顺序是对的，但进程一崩就可能丢消息；只做第三条，提交是安全的，但处理顺序仍然是乱的。

按 key 路由和按分区路由的区别很小：按 key 需要自己算哈希，按分区直接用 record.partition()。但按 key 更灵活——一个分区里可以有多个 key，路由粒度更细；按分区的并行度则被分区数钉死。

手动 assign 的代价

模型 A 和 C 都用 assign，没有走 subscribe。这带来两个直接后果：

• 没有再平衡。分区归属由代码决定，不会因为别的实例加入而改变。启动快、没有 rebalance 停顿，适合「分区数固定、实例数固定」的消费组。
• 扩容要自己做。加一个实例不会自动分走分区，你得在外部算出「哪个实例负责哪些分区」，再分别 assign。这也是为什么很多团队最后又回到了 subscribe。

assign 之后仍然可以提交位移，但 group.id 必须配置。不配的话，commitSync 会抛 InvalidGroupIdException：

org.apache.kafka.common.errors.InvalidGroupIdException:
To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.

也就是说，手动分配只是「不参与分区分配」，并没有脱离消费组。位移仍然提交到 __consumer_offsets，监控仍然能按组看 lag。

两个常见的实现错误

错误一：poll 线程自己也处理。 有人为了「不浪费」，让拉取线程也处理一部分记录：

ConsumerRecords<String, String> rs = c.poll(Duration.ofMillis(200));
for (ConsumerRecord<String, String> r : rs) {
    if (counter.getAndIncrement() % 10 == 0) {
        handle(r);              // 拉取线程自己处理
    } else {
        pool.submit(() -> handle(r));
    }
}

这段代码的吞吐通常比纯提交更低。拉取线程被处理工作占住的那段时间里，没有任何线程在发 fetch 请求；本来靠「拉取与处理重叠」换来的收益，被这个 10% 吃掉了。要么拉，要么算，别在一条线程上混着做。

错误二：worker 数拍脑袋。 共享池的线程数不是越多越好：

• 下游是 CPU 计算：worker 数 ≈ 核数即可，多出来的线程只在队列上排队；
• 下游是阻塞 IO：worker 数可以远大于核数，但要有超时和上限，否则积压会从 Kafka 转移到线程池的队列里；
• 下游是数据库：worker 数受连接池大小约束，超过连接池的部分只是在等连接。

模型 A 的并行度由分区数决定，模型 B/C 的并行度由 worker 数决定，后者可以超过分区数，但超过之后同一条分区会有多条线程在跑，顺序就没了。要「既要并行度超过分区数、又要保序」，只能按 key 再切一层路由。

选型表

上线前的检查清单

• 每个 KafkaConsumer 只被一条线程使用，创建、poll、commit、close 都在同一条线程上；
• 工作池模型提交的是「连续完成」的位移，不是 poll 到的位置；
• 关闭顺序是：停止拉取 → 等工作队列排空 → 最后一次 commitSync → close；
• max.poll.interval.ms 大于「一次 poll 的最坏处理时间」，否则会被踢出组（12.2 会实测这个过程）；
• 线程数超过分区数时，多出来的 worker 会抢同一分区的记录，顺序就断了；要并行度超过分区数又保序，只能按 key 再切一层路由；
• 记录线程数与消费者实例数，别用 JVM 的 peakThreadCount 当指标——它包含启动期的线程，参考价值有限。

回到 libraryd

借阅投影的消费组最后选了模型 A：4 个分区、4 个消费者线程，每个线程独立处理并提交。理由是投影更新是纯内存操作，没有阻塞 IO，池化只会增加交接成本。

通知服务则用了模型 C：它要按 bookId 保序地发「预约到书」提醒，同时每条提醒要走一次 HTTP，必须并发。按分区路由到 lane，既保住了同一本书的顺序，又让多个 HTTP 调用并行。

下一节处理模型 A/C 都绕不开的一个细节：提交。异步提交看起来最省事，但它丢位移的时候，调用方往往一无所知。

阅读导航：上一节 11.3 大消息的 claim-check 模式 ｜ 下一节 12.2 异步提交的陷阱