命令手册

外观

消费组与再均衡：为什么会重复消费 ​

消费组解决两件事：并行消费和位移管理。代价是再均衡——每次成员变化都会暂停消费、重新分配分区，这段时间最容易出重复和超时。

场景 ​

  • 消费慢，想加实例。
  • 日志里频繁出现 Revoking previously assigned partitions。
  • 明明处理了，重启后又收到同样的消息。

一、模型 ​

text
topic order (6 分区)
  └── group order-group
        ├── consumer-1  → 分区 0,1
        ├── consumer-2  → 分区 2,3
        └── consumer-3  → 分区 4,5

两条铁律：

  1. 同一分区在同一时刻只能被同组的一个消费者消费。
  2. 消费者数 > 分区数时，多出来的消费者完全空闲（不是分担负载）。

不同 group 之间互不影响，各自维护一份位移。

bash
/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.0.0.31:9092 \
  --describe --group order-group --state
text
GROUP                COORDINATOR (ID)          ASSIGNMENT-STRATEGY  STATE           #MEMBERS
order-group          10.0.0.32:9092 (2)         range                Stable          3

状态机：

状态含义
Empty组存在但没有成员
PreparingRebalance有成员变动，等所有人重新加入
CompletingRebalance等 leader 下发分配方案
Stable正常工作
Dead组被删除

PreparingRebalance 期间所有消费者停止消费。这就是"一加机器就全组卡几秒"的原因。

二、三个超时参数 ​

参数默认值管什么
session.timeout.ms45000多久没心跳就认为成员死了
heartbeat.interval.ms3000心跳间隔，通常是 session 的 1/3
max.poll.interval.ms300000两次 poll() 之间允许的最大间隔

前两个是心跳（后台线程自动发），第三个是业务处理时间。

max.poll.interval 超时是最常见的 rebalance 原因

处理一批消息超过 5 分钟 → 协调器认为你死了 → 踢出组 → 触发 rebalance → 你处理完回来发现分区没了 → 提交位移失败 → 消息重复。

三个解法：

  1. 减小 max.poll.records（默认 500，降到 100 或更小）。
  2. 加大 max.poll.interval.ms（但要考虑真卡死时的恢复时间）。
  3. 把耗时处理异步化：poll() 拿到就快速入本地队列，另起线程池处理——此时必须自己控制提交位移，不能用自动提交。

session.timeout.ms 不要设太小。3 台 broker 的集群，一次 Full GC 停顿 20 秒很正常，设 10 秒就会误判成员死亡。

三、再均衡的触发条件 ​

触发说明
消费者加入扩容、重启
消费者离开停机、崩溃、session 超时
max.poll.interval 超时处理太慢
订阅的 topic 变化subscribe() 参数变了
分区数变化加分区
协调器迁移该 group 的 coordinator broker 挂了

四、eager vs cooperative ​

策略行为影响
RangeAssignor（默认，eager）全部撤销再重新分配整个组暂停
CooperativeStickyAssignor只撤销需要迁移的分区大部分消费者不中断

开启协作式：

java
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
yaml
# Spring Boot
spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

切换策略要滚动重启两次

协作式和老协议不兼容。组里同时存在两种策略时，Kafka 会选择双方都支持的 RangeAssignor，等于没切。

正确流程：先全部滚动重启一次（仍在 Range）→ 再改配置全部滚动重启一次（此时都用 Cooperative）。中间会有两次 rebalance，安排在低峰。

五、静态成员：避免重启触发 rebalance ​

java
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "order-consumer-1");

group.instance.id 让协调器记住"这个成员只是短暂离线"。在 session.timeout.ms 内重启，不触发 rebalance，直接拿回原分区。

要求：

  • 每个实例的 ID 必须唯一且稳定（用 Pod 名或配置项，不要用随机值）。
  • 只适合滚动重启，不适合弹性伸缩（新实例的 ID 不在名单里，照样触发 rebalance）。

六、位移提交的三种姿势 ​

java
// 1. 自动提交（默认）——省事，但不可控
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

// 2. 同步提交——处理完再提交，可靠但慢
consumer.commitSync();

// 3. 异步提交——快，但失败时不知道丢了哪条
consumer.commitAsync((offsets, e) -> {
    if (e != null) log.error("commit failed", e);
});

生产推荐：

java
try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> r : records) {
            handle(r);
        }
        consumer.commitSync(Duration.ofSeconds(5));
    }
} finally {
    consumer.commitSync();     // 停机前最后提交一次
    consumer.close();
}

如果批次大、提交频繁，用"异步提交 + 定期同步"：

java
long lastCommit = System.currentTimeMillis();
for (ConsumerRecord<String, String> r : records) {
    handle(r);
    if (System.currentTimeMillis() - lastCommit > 1000) {
        consumer.commitSync();
        lastCommit = System.currentTimeMillis();
    }
}
consumer.commitAsync();

提交了但没处理完 = 丢消息

commitSync() 提交的是下一次要读的位置。如果先提交后处理，处理失败时这条消息永远不会再来。顺序必须是：拉取 → 处理 → 提交。

七、重复消费的四个真实原因 ​

1. 处理完、提交前崩溃 ​

text
poll → handle(100 条) → commit ← 在这一步之前进程被杀

重启后从上次提交位置开始，这 100 条重新消费。这是 at-least-once 的固有代价，只能靠业务幂等解决。

2. rebalance 时分区被抢走 ​

协作式能减少影响，但迁移的分区仍会回退到上次提交位置。正在处理的那几条会重复。

对策：处理时间不要超过 max.poll.interval，尽量让单批处理时间短。

3. 自动提交的滞后期 ​

auto.commit.interval.ms=5000，且在 poll() 里提交。你处理完的这批还没提交，下一批就来了；一次 rebalance 就会回退。

4. 位移动了但下游失败 ​

Kafka 提交成功、写数据库失败、事务回滚 —— Kafka 侧已经认为消费完了。

对策：consume → process → commit 必须放在同一个业务事务里，或者用幂等写。

sql
-- 幂等写：重复消费不会产生重复数据
INSERT INTO orders (id, amount) VALUES (?, ?)
ON CONFLICT (id) DO NOTHING;

八、消费者配置模板 ​

yaml
spring:
  kafka:
    bootstrap-servers: 10.0.0.31:9092,10.0.0.32:9092,10.0.0.33:9092
    consumer:
      group-id: order-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      max-poll-records: 100
      properties:
        session.timeout.ms: 45000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000
        isolation.level: read_committed
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
    listener:
      ack-mode: manual_immediate
      concurrency: 3

spring.kafka.listener.concurrency 是每个实例起的消费线程数。总并行度 = 实例数 × concurrency，应 ≤ 分区数。

九、优雅停机 ​

停机时不 close() 会导致：

  • 位移没提交，重启后重复消费。
  • 协调器要等 session.timeout 才发现你走了，期间 rebalance 卡住。
java
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    running = false;
    consumer.wakeup();          // 打断正在进行的长 poll
}));

Spring Boot 里：

yaml
spring:
  lifecycle:
    timeout-per-shutdown-phase: 30s
server:
  shutdown: graceful
java
@KafkaListener(topics = "order", groupId = "order-group")
public void onMessage(ConsumerRecord<String, String> r, Acknowledgment ack) {
    handle(r);
    ack.acknowledge();
}

ack-mode: manual_immediate + ack.acknowledge() 表示"这条处理完立刻提交"。要比 record 模式更精确。

十、踩过的坑 ​

现象根因处置
加了实例但吞吐没涨实例数 > 分区数扩分区（注意 key 顺序）
日志反复出现 rebalance处理超 max.poll.interval减小 max.poll.records 或异步化
每次发版重复消费几分钟的量停机没提交位移加 shutdown hook，优雅停机
消费组一直是 PreparingRebalance有成员起不来/一直超时看 --describe --state --members
重启后从头消费auto.offset.reset=earliest 且组没有提交记录确认 group 名没变，或先提交一次
部分消费者拿不到分区分配策略不均 / 消费者数超过分区换 CooperativeStickyAssignor
CommitFailedExceptionrebalance 期间提交处理完成后立即提交，缩短处理时间
改了策略但没生效组内策略不一致，回退到 Range分两次滚动重启