消费者组

这一课你将能做到

1. 配置 group.id、enable.auto.commit=false、auto.offset.reset，用 Java Consumer 拉取消息并 处理成功后再 commitSync
2. 画出「分区数 vs 实例数」与 rebalance 时暂停消费的图像，说明为什么处理逻辑要快且幂等
3. 对照 Kafka Consumer Group 与 Redis Pub/Sub，说出离线补消息、竞争消费、进度记录三点差异

为什么要学

Producer 把事件写进分区后，真正产生业务副作用的是 Consumer：发通知、写审计、刷搜索索引。Consumer Group 决定「谁来消费、消费多少、消费到哪」：

• 同一组内多实例分摊分区——这就是 Kafka 的水平扩展方式（部署两个实例吞吐翻倍的前提是分区 ≥ 2）
• 不同组各自完整消费——通知和审计都要全量事件，互不干扰

默认工程语义接近 at-least-once：可能重复，必须幂等。本课把位点提交、rebalance、poll 循环三件事钉死，它们同时是 Kafka 面试出现频率最高的三个问题。

人话：一组读者共用一本日记的不同页（组内分摊）；另一组读者有自己的书签，可以从头再读一遍全书（组间独立）。

先建立图像

Topic order-events   P0  P1  P2
                      │
        ┌─────────────┼─────────────┐
        ▼                            ▼
 group notify (负载均衡)        group audit (再订阅)
  实例1 → P0, P1                 实例X → P0, P1, P2（可以一个人扛）
  实例2 → P2                     各组独立维护 CURRENT-OFFSET

Rebalance 触发：实例上线/下线/崩溃、订阅变更
  → 组内暂停拉取 → 重新分配分区 → 从各组已提交位点继续

	Kafka Consumer Group	Redis Pub/Sub
离线后补消息	可从未提交/earliest 位点读	收不到历史
多实例	竞争分摊分区	广播给所有订阅者
消费进度	Offset（组 × 分区）	无

跟着做

步骤 1：准备 Topic 与假数据

确保 coderoadmap.order-events 存在。用 console producer 先灌 5 条：

docker exec -it kafka-lab /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic coderoadmap.order-events

本课用一个 新的 group.id（如 coderoadmap-notify-lab），避免和旧实验位点搅在一起。

步骤 2：写手动提交 Consumer

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class OrderEventConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 同组多进程分摊分区；换组名 = 新订阅者
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "coderoadmap-notify-lab");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关闭自动提交：避免"位点推了但业务没做完"
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 无已提交位点时才生效：earliest 从最早读，latest 只吃新消息
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 心跳与拉取节奏（面试常问，先认识）
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "45000");
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("coderoadmap.order-events"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    System.out.printf("partition=%d offset=%d key=%s value=%s%n",
                            r.partition(), r.offset(), r.key(), r.value());
                    handleOrderEvent(r.value()); // 业务处理：必须幂等
                }
                if (!records.isEmpty()) {
                    consumer.commitSync(); // 全部处理成功后再提交位点
                }
            }
        }
    }

    static void handleOrderEvent(String json) {
        // 实验：打印即可；生产要有幂等（去重表），见"投递语义"课
    }
}