消费者与消费组

一句话：分区是并行单位，重平衡是分配单位——一个分区同一时刻只能归组内一个消费者，组内并行度上限就是分区数。

适用版本：Kafka 3.9 / 4.0。默认值以 Java 客户端为准；KIP-848 新协议在 3.7 预览、4.0 GA。

为什么需要

没有消费者组时，你要自己决定每个消费者读哪些分区，还要自己处理宕机后的接管。分区一多、消费者一扩缩容，手工分配立刻失控。

消费者组把这件事变成协议：成员通过 Coordinator 协商「谁读哪些分区」，成员变化时自动重分配，位点集中存在 __consumer_offsets 里，重启后接着读。代价是重平衡期间消费暂停（协作式只暂停被移动的分区），且分区数一旦定下，组内并行度就封顶。

核心机制

角色与位点

Coordinator 是某个 Broker 上的组件，负责成员管理和重平衡；位点写入内部主题 __consumer_offsets（默认 50 个分区）。Coordinator 从成员里选一个消费者当 Group Leader，由它计算分配方案；Group Leader 不是 Broker 的 Leader。每个 <group, topic, partition> 记录一个已提交 offset，重启或换成员后从这里继续。

JoinGroup / SyncGroup

一次经典重平衡分两轮：所有成员发 JoinGroup，带上订阅的 Topic 和分配策略；Coordinator 选一个成员当 Group Leader，把成员列表返回给它；Leader 按策略算出「成员 → 分区」方案，通过 SyncGroup 提交；Coordinator 把结果下发给所有成员，进入 Stable 状态。

触发条件：成员加入/主动离开、订阅变化、分区数增加、session.timeout.ms 内没有心跳、单次 poll() 处理超过 max.poll.interval.ms，以及消费组协议升级。

分配策略

策略	行为	代价
Range	按 Topic 把连续分区分给消费者	多 Topic 时容易不均
RoundRobin	所有 Topic 的分区排序后轮询	每次重平衡全量重排
Sticky	尽量均匀，且尽量少移动已有分区	仍是 eager，先全部停再分配
CooperativeSticky	增量撤销/分配，只暂停要移动的分区	需要客户端支持，不能和 eager 混用

Kafka 3.x 的 partition.assignment.strategy 默认是 RangeAssignor, CooperativeStickyAssignor 的组合，按所有成员都支持的第一个生效；不同版本可能调整，以官方文档为准。新项目显式配 CooperativeStickyAssignor 可以减少重平衡抖动。

offset 提交的三种语义

语义	做法	结果
at-most-once	先提交 offset，再处理消息	处理失败就丢消息
at-least-once	先处理，成功后再提交	崩溃/重平衡会重复
exactly-once	处理结果与 offset 放进同一个 Kafka 事务	仅覆盖 Kafka 内部读写，外部副作用仍要幂等

自动提交 enable.auto.commit=true（默认）每 auto.commit.interval.ms=5000 提交一次，提交的是「poll 到的位置」，不是「处理完的位置」，所以既可能丢也可能重。需要可控语义时关掉自动提交，处理完一批再 commitSync()；commitAsync() 不阻塞，但失败回调要自己处理。

找不到位点时按 auto.offset.reset 决定：latest（默认）从末尾读，earliest 从最早可用位置读，none 直接抛异常。位点保留 offsets.retention.minutes=10080（7 天），组长期不消费后会被清理。

max.poll.interval.ms 与 max.poll.records 的因果

心跳由后台线程发，但「成员是否活着」还有另一条判据：两次 poll() 的间隔不能超过 max.poll.interval.ms（默认 300000，5 分钟）。

poll() 一次最多返回 max.poll.records（默认 500）条。如果单条处理 100ms，一批就是 50s；加上下游抖动，很容易超过 5 分钟，Coordinator 会把这个成员踢出组，触发重平衡。重平衡后分区可能换人，已处理但没提交的消息会被重新消费。

所以这两个参数是一对预算：max.poll.records × 单条处理耗时 < max.poll.interval.ms。超了只有两个方向：调小 max.poll.records，或把处理挪到异步线程并让 poll 线程持续调用 poll()。单纯调大 max.poll.interval.ms 只是把故障推迟。

静态成员

group.instance.id（2.3+）给成员一个固定身份。滚动重启时，新实例在 session.timeout.ms 内用同一个 ID 重新加入，Coordinator 认为还是原成员，不触发重平衡。前提是同一时刻不能有两个实例用同一个 ID，否则会互相 fencing。它适合 StatefulSet/固定实例名的部署，不适合随时扩缩容的无状态副本。

lag 的观测口径

LAG = LOG-END-OFFSET - CURRENT-OFFSET

kafka-consumer-groups.sh --describe 的输出里：CURRENT-OFFSET 是已提交位点，LOG-END-OFFSET 是分区当前末尾（LEO），LAG 是还差多少条，CONSUMER-ID/HOST/CLIENT-ID 是当前成员；为空或 - 说明这个分区没有活跃消费者。

lag 是快照：它不区分「正在处理」和「还没拉取」，也不代表消息一定没被业务处理。判断积压要看 lag 的趋势，以及 --members --verbose 里的分区归属。

实战场景

一、lag 归零，消息却没被处理

现象：营销标签消费者用了默认配置（enable.auto.commit=true）。某次发布后进程被 kill，重启后发现约 4800 条用户事件没有落到标签表，但 --describe 显示 LAG=0，监控一片绿，最初误判为“下游写库丢了数据”。

排查：自动提交的是“poll 到的位置”，不是“处理完的位置”。

定位层	取证据	判定
提交语义	`auto.commit.interval.ms=5000` 期间是否 poll 了但未处理完	poll 后 5s 就提交，处理失败也提交
崩溃点日志	是否出现 `poll` 返回后异常、但没有 commit 记录	异常发生在处理中、位点已推进即命中
位点与数据对账	按事件时间窗统计“应有条数”与“实落条数”	差值与 lag 无关，只能靠对账发现
失败处理	处理异常是否只 `log.warn` 后继续	吞异常 + 自动提交 = 最多一次语义
重置位点影响	是否曾用 `--to-latest` “消积压”	latest 会跳过未消费数据，掩盖问题

对症做法：

java

// 1. 需要“至少一次”就关掉自动提交，处理完整批再同步提交
props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    if (records.isEmpty()) continue;
    try {
        process(records);                       // 全部处理成功
        consumer.commitSync();                  // 位点与处理结果一致
    } catch (RetryableException e) {
        retryLater(records, e);                 // 不提交：本批会被重新消费，要求处理幂等
        metrics.retry("label");
    } catch (FatalDataException e) {
        for (ConsumerRecord<String, String> r : records) dlq.send(r, e);   // 脏数据进死信，别堵整批
        consumer.commitSync();
    }
}

text

2. 处理必须幂等：至少一次意味着同一批可能被处理两次，
   标签写入用 (userId, tagId) 唯一键或 UPSERT 收敛。
3. 单条脏数据不能拖住整批：要么 per-record 提交位点（吞吐换可控），
   要么进死信后继续，绝不允许“跳过并继续提交”却不留记录。
4. 真正的验收靠对账：按事件时间窗比较源侧条数与目标侧条数，
   lag 只能看趋势，不能证明业务处理完。

复盘要点：判据是“LAG=0 不作为完成信号”——补一个业务口径指标“窗口内应有/实落比值”，低于 1 就告警。语义与配置的对应要写进代码模板：自动提交=最多一次倾向、手动提交=至少一次、事务=Kafka 内精确一次，三选一必须在评审时明确。

二、一次下游抖动引发重平衡风暴

现象：风控服务响应从 80ms 涨到 2s，标签消费者开始反复进出组：日志里 Attempt to heartbeat failed since member has left、Rebalance started 每隔几分钟一轮，lag 从 2 万涨到 40 万，恢复后仍要继续消化很久。

排查：心跳有后台线程，但“成员是否活着”还看两次 poll() 的间隔。

定位层	取证据	判定
单批耗时预算	`max.poll.records × 单条处理耗时` vs `max.poll.interval.ms`（默认 300000）	500 × 2s = 1000s 远超 5 分钟，必然被踢
成员变动	`--members --verbose` 里的成员 ID 与加入时间是否频繁变化	反复进出即重平衡风暴
位点是否提交	被踢时未提交的批次	重平衡后从头再处理，重复消费放大压力
分配策略	是否仍带 eager 策略	eager 会全体停下，风暴期间全组暂停
会话超时	`session.timeout.ms=45000`、`heartbeat.interval.ms=3000`	单纯调大 session 不解决处理慢
下游容量	消费者重启与下游 QPS 曲线	重复消费把下游打得更快，正反馈

对症做法：

properties

# 1. 把批次调小，让单轮处理稳稳低于预算（这里目标 < 60s，留 5 倍余量）
max.poll.records=50
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
# 2. 固定实例名部署（StatefulSet）用静态成员，滚动重启不触发重平衡
group.instance.id=label-worker-3

java

// 3. 慢处理挪出 poll 线程，poll 线程只负责持续调用 poll()
ExecutorService biz = new ThreadPoolExecutor(8, 8, 60, SECONDS,
        new ArrayBlockingQueue<>(maxInFlight), new CallerRunsPolicy());  // 队列满时反压 poll 线程

while (true) {
    ConsumerRecords<String, String> rs = consumer.poll(Duration.ofMillis(200));
    drainCompleted();                       // 回收已完成任务并提交位点（按分区取最小 offset+1）
    if (biz.getQueue().remainingCapacity() < 10) { Thread.sleep(20); continue; }  // 反压
    for (ConsumerRecord<String, String> r : rs) submitTracked(r, biz);
}

text

4. 位点管理变复杂后的更稳做法：处理完写“业务处理完成标记”，
   按 <partition, minCompletedOffset+1> 提交，避免乱序完成导致位点越过未完成的消息。
5. 不要用调大 max.poll.interval.ms 收尾：那只是把故障推迟；
   真正的约束是“单轮处理时间必须显著小于该值”。

复盘要点：判据是“成员进出次数应为 0（除发布窗口）”，把 rebalance-total（客户端指标）做成趋势告警。容量口径要写清楚：组内真正干活的消费者不超过分区数，所以下游慢的时候先降批次、再加下游容量，最后才考虑扩分区——扩分区会改 key 落点，顺序性保证随之失效，见架构与 ISR。

三、热点 key 让单分区 lag 40 万，其他分区空闲

现象：订单状态 Topic 12 个分区，其中一个分区 lag 长期 40 万，其余接近 0；把消费者从 12 个扩到 24 个没有任何改善，反而有 12 个成员完全空闲。

排查：并行度封顶在分区数，且分区内并行度由 key 分布决定。

定位层	取证据	判定
消费者与分区比	组内成员数 vs Topic 分区数	成员多于分区，多出来的一定向空闲
分区归属	`--members --verbose` 看每个分区挂在哪个成员	一个成员扛住全部高 lag 分区即可
key 分布	按 `hashCode(key) % 12` 统计近 1 小时消息落点	某分区占比 > 30% 即倾斜（经验阈值，需校准）
热点来源	具体大商家/大活动的主键	单 key 自身无法靠分区内并行拆分
处理耗时	该分区单条处理时间是否也更高	热点常伴随重处理逻辑，双重放大

分区数只能增不能减，一个分区同一时刻只能归组内一个消费者——这是“加机器解决不了”的根本原因。

对症做法：

java

// 1. 先扩消费者到分区数（12 个），超出部分只是空闲，不改善倾斜
// 2. 热点 key 用“二次分发”拆成多个子分区键，代价是放弃该 key 的严格顺序
String routeKey = isHotSeller(sellerId) ? sellerId + "#" + (threadLocalRand(8)) : sellerId;
producer.send(new ProducerRecord<>("order-status", routeKey, payload));

// 3. 需要保留顺序时改成“同 key 串行、不同 key 并行”的本地处理
//    按 key 哈希到固定线程，保证同 key 单线程处理，跨 key 并行
Executor[] lanes = new Executor[32];
int lane = Math.abs(key.hashCode()) % lanes.length;
lanes[lane].execute(() -> handle(record));

text

4. 扩分区要挑消费端 lag=0 的窗口，且明确该 key 空间不再跨新旧分区有序；
   顺序敏感的 Topic 更稳的方案是“新建 Topic + 双写 + 切读”，
   分区数按 3 年容量一次定足。
5. 热点是业务现象，技术兜底之外要有业务侧限流：单个大商家的批量操作排队下发，
   不要把瞬时并发直接推给 Kafka。

复盘要点：判据是“单分区 lag 占比”——最热分区 lag / 总 lag 长期超过 50% 就说明 key 设计与并行度不匹配，光加消费者无效。上线前的容量测算要按“预估峰值消息数 ÷ 分区数 = 单分区需要承接的速率”倒推分区数，而不是照抄默认值；位点保留期 offsets.retention.minutes=10080（7 天）也要注意长期停用的组会被清位点，重启后会按 auto.offset.reset 决定起点，这一条在灾备演练里最容易踩。

最小上手

bash

# 1. 建一个 3 分区 Topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 3 --replication-factor 1

# 2. 起一个带 group 的消费者（另开终端）
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --group order-worker \
  --property print.offset=true --property print.partition=true

# 3. 生产几条消息
printf 'o-1\no-2\no-3\n' | bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic orders

# 4. 看位点与 lag；列顺序：GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
#    看成员与分区归属时追加 --members --verbose
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-worker

重置位点必须先停掉消费者，且只能对非活跃组执行。--dry-run 先看影响范围，--execute 才生效；earliest 会重放，latest 会跳过未消费数据：

bash

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-worker --topic orders --reset-offsets --to-earliest --dry-run
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-worker --topic orders --reset-offsets --to-earliest --execute

删除组会一并删掉已提交位点，组必须已无活跃成员：

bash

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --delete --group order-worker

生产动作

Java 消费者侧最该先定死的参数：

配置	默认值（3.9/4.0）	说明
`enable.auto.commit`	`true`	需要至少一次语义时改 `false`
`auto.commit.interval.ms`	`5000`	自动提交周期
`auto.offset.reset`	`latest`	无位点时的起点
`max.poll.records`	`500`	单次 poll 最大条数
`max.poll.interval.ms`	`300000`	两次 poll 的最大间隔
`session.timeout.ms`	`45000`	3.0 起默认 45s，之前是 10s
`heartbeat.interval.ms`	`3000`	3.0 起默认 3s
`group.instance.id`	`null`	静态成员 ID
`partition.assignment.strategy`	Range + CooperativeSticky	以官方文档为准
`isolation.level`	`read_uncommitted`	事务场景改 `read_committed`，见 04

排障顺序：

--describe 看 lag 是单分区还是全部分区。单分区先查 key 倾斜，全部分区先查消费者数量和下游。
--members --verbose 看分区是否集中在少数成员，或有没有成员反复进出；消费者日志搜 Rebalance、Attempt to heartbeat failed、member has left。
lag 持续增长时先扩消费者到分区数，再考虑扩分区；扩分区会改变 key 的落点，顺序性保证随之失效。

减少重平衡的常用组合：CooperativeStickyAssignor + 固定 group.instance.id + 单批处理时间稳稳低于 max.poll.interval.ms。不要用调大 session.timeout.ms 掩盖处理慢：心跳线程还活着时，处理慢由 max.poll.interval.ms 判定。

边界

并行度封顶在分区数。消费者多于分区时，多出来的成员空闲，不会提升吞吐。
重平衡不是零成本。协作式只减少暂停范围，不消除暂停；eager 策略仍是全体先停。
自动提交不等于处理成功。lag=0 只说明位点追上了，不能说明业务处理完了。
重置位点不是积压的解药。--to-latest 会跳过未消费数据，--to-earliest 会引发重复消费和下游压力。先定位为什么消费不动。
静态成员有 fencing 风险。同一个 group.instance.id 同时起两个实例，后加入的会把前一个踢掉。
投递语义不在本篇。acks、幂等、事务见 02-Producer与投递语义；3.0 起 acks 默认已是 all、enable.idempotence 默认已是 true，不要再按旧默认值理解。
KIP-848 是另一套协议。3.7 预览、4.0 GA，客户端用 group.protocol=consumer 显式启用；启用后 session.timeout.ms、heartbeat.interval.ms、partition.assignment.strategy 由 Broker 侧控制，且不支持自定义分配器。4.0 客户端默认仍是 classic，细节以官方文档为准。

讲清楚

消费者组只需要记住两句话：分区是并行单位，重平衡是分配单位。一个分区同一时刻只能归组内一个消费者，所以组内最多有「分区数」个消费者真正干活，多出来的空闲。
重平衡是成员和 Coordinator 协商「谁读哪些分区」的过程：所有人 JoinGroup，Coordinator 选一个消费者当 Leader，Leader 按策略算方案，SyncGroup 下发。成员增减、订阅变化、心跳超时、单次 poll 处理超过 5 分钟都会触发。eager 策略先全部停下再分配，协作式只动需要移动的分区。位点提交决定语义：先提交后处理是最多一次，先处理后提交是至少一次，把处理结果和位点放进同一个 Kafka 事务才是 Kafka 内的精确一次。
线上最该盯的是 lag 趋势，以及 max.poll.records × 单条处理耗时 是否超过 max.poll.interval.ms。超了就会被踢出组、触发重平衡、重复消费。先调小批量或异步处理，再谈扩消费者；扩消费者也不能超过分区数。

Kafka 索引

RocketMQ

ElasticJob

ZooKeeper

XXL-JOB

消费者与消费组

为什么需要

核心机制

角色与位点

JoinGroup / SyncGroup

分配策略

offset 提交的三种语义

max.poll.interval.ms 与 max.poll.records 的因果

静态成员

lag 的观测口径

实战场景

一、lag 归零，消息却没被处理

二、一次下游抖动引发重平衡风暴

三、热点 key 让单分区 lag 40 万，其他分区空闲

最小上手

生产动作

边界

讲清楚

消费者与消费组 ​

为什么需要 ​

核心机制 ​

角色与位点 ​

JoinGroup / SyncGroup ​

分配策略 ​

offset 提交的三种语义 ​

max.poll.interval.ms 与 max.poll.records 的因果 ​

静态成员 ​

lag 的观测口径 ​

实战场景 ​

一、lag 归零，消息却没被处理 ​

二、一次下游抖动引发重平衡风暴 ​

三、热点 key 让单分区 lag 40 万，其他分区空闲 ​

最小上手 ​

生产动作 ​

边界 ​

讲清楚 ​

消费者与消费组

为什么需要

核心机制

角色与位点

JoinGroup / SyncGroup

分配策略

offset 提交的三种语义

max.poll.interval.ms 与 max.poll.records 的因果

静态成员

lag 的观测口径

实战场景

一、lag 归零，消息却没被处理

二、一次下游抖动引发重平衡风暴

三、热点 key 让单分区 lag 40 万，其他分区空闲

最小上手

生产动作

边界

讲清楚