Consumer 与消费组

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
 
public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "inventory-service");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 手动提交
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(List.of("orders"));   // 订阅后并未立刻分配分区
 
        try {
            while (true) {
                // poll 做三件事: 心跳协调、拉取消息、触发 Rebalance 回调
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    System.out.printf("p=%d offset=%d key=%s value=%s%n",
                        r.partition(), r.offset(), r.key(), r.value());
                    // 业务处理...
                }
                consumer.commitSync();   // 处理完一批再提交 (第 8 章详解)
            }
        } finally {
            consumer.close();   // 主动离组, 立刻触发 Rebalance 而非等超时
        }
    }
}

from confluent_kafka import Consumer
 
consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "inventory-service",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"])
 
try:
    while True:
        msg = consumer.poll(timeout=0.5)    # 一次一条 (也可用 consume 批量)
        if msg is None:
            continue
        if msg.error():
            print(f"error: {msg.error()}")
            continue
        print(f"p={msg.partition()} offset={msg.offset()} value={msg.value()}")
        consumer.commit(asynchronous=False)
finally:
    consumer.close()

1. FindCoordinator  找到自己组的协调器
2. JoinGroup        入组; 协调器选一个消费者当 "组长" (group leader)
3. 组长根据分配策略计算 分区->消费者 的映射
4. SyncGroup        组长上报方案, 协调器下发给所有成员
5. 各消费者开始 poll 自己名下的分区, 周期性发心跳

线 1: 心跳线程 (后台自动)                线 2: 你的业务处理 (poll 间隔)
 ├ heartbeat.interval.ms  发心跳周期      └ max.poll.interval.ms
 └ session.timeout.ms     判死时限            两次 poll 之间的最大间隔
   "进程还活着吗?"                            "还在正常处理消息吗?"

症状: 消费组反复 Rebalance, 日志出现
  "consumer poll timeout has expired ... max.poll.interval.ms"
原因: 单批消息处理时间 > max.poll.interval.ms
解法 (按优先级):
  1. 减小 max.poll.records (默认 500), 一批少拉点
  2. 加快单条处理 (异步化、批量写库)
  3. 调大 max.poll.interval.ms (治标)

session.timeout.ms=45000
heartbeat.interval.ms=15000        # 一般设为 session 的 1/3
max.poll.interval.ms=300000
max.poll.records=500               # 每批最大条数, 按单条处理耗时倒推
fetch.min.bytes=1                  # broker 攒够多少字节才返回
fetch.max.wait.ms=500              # 攒不够时最多等多久

// 手动指定消费 orders 的 0 号分区
TopicPartition p0 = new TopicPartition("orders", 0);
consumer.assign(List.of(p0));
consumer.seekToBeginning(List.of(p0));