消费组与协作

消费组解决两个问题：⁠谁负责处理哪个分区，以及换人之后从哪里继续⁠。前者是所有权协调，后者是进度持久化。只有两者配合，多个进程才能对同一条逻辑消费任务分工并恢复。

本文讨论 Kafka 的常规 consumer group，不包含 share group。先读 Kafka 架构 可了解 partition 与 offset；处理重复和外部副作用的保证见 消息语义。

分组意味着共享任务，不意味着共享读取位置

假设 topic 有六个分区，库存组有三个成员：

同一组内，每个分区分配给一个成员；不同组各自读取同一日志。G1 与 G2 的进度相互独立，不是一个组先读完，另一个组才有资格读。

六个分区最多分配给六个持有分区的组成员，第七个成员可能闲置。增加分区能提高分片并行度，但也改变 key 映射与状态分布。分区内部可以使用工作线程并发执行，不过顺序完成、状态竞争和 offset 提交就变成应用的责任。

Coordinator、成员身份与分配

组协调者管理成员、分配与进度相关协议。成员通过心跳等交互证明仍然参与组，协调者在成员离开、超时或订阅变化后安排新的分配。

这里应区分两个层次：broker 侧如何可靠保存组状态，和消费者如何获得分区分工。消费者并不是彼此执行一轮 Raft 来达成分配共识；把消费组笼统称作“共识算法”会掩盖实际协议角色。

成员身份还承担隔离旧实例的作用。一个进程失联后可能仍在运行，不能仅因为新成员已经接手，就假设旧进程停止了外部数据库写入。Kafka 能约束自己的组协议与提交，外部副作用仍需事务、幂等或业务 fencing。

再平衡：安全交接比均匀分配更重要

以 P1 从 C1 转交 C4 为例，安全交接至少要处理这些状态：

如果 C1 来不及提交就崩溃，C4 重放最近一段是正常恢复行为。如果 C1 已失去所有权，不能把旧任务完成后的 offset 当成仍有权提交；同样不能让其在外部存储无限写入。

再平衡的代价包括停止服务的时间、缓存预热、状态恢复和重放，不只是重新计算一张分配表。

Revoked 与 Lost 不能执行完全相同的收尾

正常撤销时，应用可以停止派发、有限等待在途任务，再尝试提交已完成水位；等待不能无限阻塞组协议。提交失败后允许重放，比提交尚未完成的位置更容易保持处理承诺。

已经丢失分区时，新成员可能开始工作，旧成员应撤销本地任务资格并清理状态。此时不能把 onPartitionsLost 当成一次仍拥有分区的最终提交机会。两种回调的差异见 ConsumerRebalanceListener。

本地资格检查只保护这个进程的派发和回调。旧任务若已经向数据库发出请求，设置一个 cancelled 标志无法收回该请求。需要外部存储在实际写入时原子检查授权代次，或按稳定事件 ID 去重；如果只在发请求之前查询一次所有权，查询后到写入前仍可能发生接管。

Eager、Cooperative 与新 Consumer 协议

“渐进交接”与“使用哪一代组协议”是两个维度，不能简单等同。按 Kafka 4.1 文档：

Classic 中的 cooperative sticky assignor 与 KIP-848 引入的 Consumer 协议不是同一个功能。协议支持、迁移约束和服务端分配器见 Consumer Rebalance Protocol。

Sticky 的目标是减少不必要移动；均衡的目标是分摊负载。分区数均匀也不一定工作量均匀：一个热点分区可能比其他十个分区都忙。仅更换 assignor 不会自动切开热点。

Offset 是恢复承诺，不是拉取计数

提交值通常表示下一条应读取的位置⁠。处理完 offset 100，提交 101。应用真正要维护的是“此前所有记录都已完成”的连续水位。

即使 103 已完成，也不能提交 104：此时崩溃会让 101 被跳过。等 101 成功后，水位才可以越过已经完成的 102、103，一次推进到 104。

如果只是取所有工作线程完成 offset 的最大值，提交会越过空洞。按分区维护未完成集合和连续水位，才能把并行执行转化为可靠恢复进度。分区间可独立推进，不需要让一个慢分区阻塞所有分区提交。

连续的是已交付任务前缀，不是每一个整数 offset

若某次 poll 对一个分区返回 [100, 103, 107]，跳过的位置可能来自压缩或事务过滤。等待 101、102 的完成通知会永远卡住；直接取已完成最大值又会越过真实未完成任务。正确依据是按读取顺序登记的任务队列⁠。

以下模型一次只接纳一个批次，批次完全完成后才接纳下一批，便于把恢复条件写清楚。next_position 是该分区这次读取之后的位置，只有全部已交付任务完成才可提交它。token 是应用本地唯一的批次资格，关联当前分区的这次分配；它不是 Kafka 协议中的 generation 字段。

class BatchProgress:
    def __init__(self, offsets, next_position, token):
        if offsets != sorted(set(offsets)):
            raise ValueError("offset 必须严格递增")
        if next_position < 0 or (offsets and next_position <= offsets[-1]):
            raise ValueError("读取结束位置必须在最后记录之后")
        self.offsets = list(offsets)
        self.next_position = next_position
        self.token = token
        self.active = True
        self.done = set()
        self.front = 0

    def completed(self, offset, token):
        if not self.active or token != self.token:
            return False  # 旧分配、旧批次或撤销后的回调
        if offset not in self.offsets:
            raise ValueError("这个批次没有派发该 offset")
        self.done.add(offset)
        while (self.front < len(self.offsets)
               and self.offsets[self.front] in self.done):
            self.front += 1
        return True

    def committable(self):
        if not self.active:
            return None
        if self.front == len(self.offsets):
            return self.next_position
        return self.offsets[self.front]  # 最早尚未完成任务的位置

    def revoke(self):
        self.active = False

old = BatchProgress([100, 103, 107], 110, token="assignment-A/batch-1")
assert old.committable() == 100
assert old.completed(107, old.token)
assert old.committable() == 100
assert old.completed(100, old.token)
assert old.committable() == 103
assert old.completed(100, old.token)  # 重复完成通知不推进两次
assert old.committable() == 103
old.revoke()
assert not old.completed(103, old.token)
assert old.committable() is None

new = BatchProgress([103, 107], 110, token="assignment-B/batch-1")
assert not new.completed(103, old.token)
assert new.committable() == 103
assert new.completed(107, new.token)
assert new.completed(103, new.token)
assert new.committable() == 110
assert BatchProgress([], 110, token="empty").committable() == 110

这里 100 完成后提交 103 是安全的，因为该次顺序读取已经确定 101、102 没有需要派发的业务记录，而 103 尚未完成。全部完成后提交 110，还能跨过读取位置已越过的控制记录；不能未经读取就凭猜测跨越 offset。真实客户端需按分区捕获读取后的 next offset，并将其与对应批次绑定，避免拿后来一次 poll 的位置替代旧批次边界。位置定义见 KafkaConsumer.position。

示例中假设 103 的恢复位置已经成功提交，新成员才从 103 接管。107 虽在旧成员执行过，仍可能重放；进度追踪防止遗漏，不消除重复效果。模型也没有调用 Kafka 提交接口：committable() 只是候选值，提交成功与否还要由拥有该分区的消费线程处理。工作线程通过完成队列交回结果，不直接并发调用同一个 KafkaConsumer。

多批次流水线可把模型扩展为每分区的有序队列，但必须保留每批的读取边界、分配身份和未完成任务；不能每次 poll 直接覆盖上一批状态。

完成了 107，为什么不能直接提交 108？

进度条应跟随已交付任务的顺序，而不是最大的完成编号。下面先完成 107，再完成 100，可提交位置会停在仍未完成的 103；全部完成后才跳到已知 next position 110。候选水位与持久水位分开显示，崩溃只使用后者恢复，所以“做过了”仍可能需要重放。

自动提交与手动提交

自动提交按客户端机制定期保存消费位置；它不知道应用线程池里的业务任务是否完成。同步拉取并完成处理，与拉取后立即异步派发，是两种不同的使用方式。

手动提交提供控制权，但不会自动消除重复：业务完成到提交成功之间仍存在崩溃窗口。异步提交还要处理请求完成顺序和失败回调，不能在失败后盲目重试旧 offset，把更新的进度覆盖回去。

当消息无法处理时，需要明确业务策略：阻塞该分区等待修复，还是持久记录失败并转移到专门通道。简单跳过再提交，实质上改变了“所有消息已成功处理”的承诺。具体 API 与线程限制见 KafkaConsumer Javadoc。

心跳、处理超时与背压

“进程活着”和“业务在推进”不是一回事。心跳帮助判断成员存活；poll 间隔等约束还用于发现无法及时参与消费协议的应用。只调大超时会延缓误判，也会延缓真正故障后的接管。

设输入速率为每秒 1000 条，处理能力为每秒 800 条，积压增长约为每秒 200 条。持续十分钟就是约 12 万条。只增大客户端缓冲不会消除这个差值，只会把积压从 broker 搬进进程内存。

背压应让拉取与处理能力匹配：限制在途任务数，必要时暂停分区拉取，同时继续正常参与组协议。批次大小、工作线程数和 poll 频率要共同设计。需要保持顺序的分区，可以在分区之间并行，而在分区内串行。

暂停阈值要留出一个已经取回的批次

设本地允许保留 1200 个任务，一次最多接纳 200 个。若等队列达到 1200 才暂停，当前已经返回的一批可能使它达到 1400。可以在在途与排队总数达到 1000 时暂停，到 400 以下再恢复；两个阈值之间的间隔避免每完成一个任务就频繁暂停和恢复。

这个预算假设派发前检查容量，并且只存在一个最多 200 条的待接纳批次。若还有多个预取批次，必须全部纳入预算。按条数限流也不等于按字节限内存：200 条大消息和 200 条小消息的内存成本不同。

Kafka 的 max.poll.records 限制一次 poll 返回的记录数，不限制底层 fetch 已缓存的数据。因此任务上限要与字节预算一起设计，暂停期间仍按客户端要求调用 poll 参与协调；不能用工作队列阻塞消费线程直到失去成员资格。参见 Consumer Configs：max.poll.records 与 max.poll.interval.ms。

Lag 应如何解释

常见 lag 是某个日志边界与已提交位置的差。这个数适合观察趋势，但不总等于待处理业务记录数，也不能直接换算成延迟：记录大小、压缩后的空洞和分区流量都可能不同。

至少要同时观察：

• 每个分区的 lag 和最老未处理事件年龄，避免总量掩盖单分区热点。
• 拉取位置与提交位置的距离，判断积压在 broker 还是应用内部。
• 处理吞吐、耗时分布、在途任务量，识别下游瓶颈。
• 再平衡频率与分区恢复时间，区分稳定慢消费和反复接管。

这些量分别描述“剩多少”“等多久”“卡在哪里”。可以与 指标与告警 中的服务目标结合。

与队列消费者的比较

RabbitMQ 的经典队列、仲裁队列侧重消息分发与确认，Kafka 常规消费组侧重日志分区的分工与回放。但 RabbitMQ 也有 Streams，不能把整个产品概括为“确认即删除、永远不能回放”。见 RabbitMQ Streams。

选型应比较具体模型：需要按事件独立确认还是按连续位置提交，需要保留历史还是完成即释放，需要多组独立回放还是多个 worker 竞争任务。Kafka 提供分区内顺序，并不提供整个 topic 的全局有序。

参考

• Kafka：Consumer Rebalance Protocol
• KafkaConsumer：线程、提交与消费模式
• RabbitMQ：Streams
• 消息语义：重复执行与业务效果

关键词：consumer group、coordinator、rebalance、cooperative、offset、连续水位、背压、lag、fencing

积压恢复与提交正确性分别验证

提交边界正确以后，积压仍可能持续增长。容量与过载恢复用到达工作量、服务余力和重试负担解释追赶时间；计算机系统中的控制再处理扩容延迟和调节振荡。增加消费者只能在分区、外部存储与处理能力允许时提高有效吞吐，不能替代本篇的持久提交与去重协议。