📚 课程目录第3章生产者与消费者基础

3. 生产者与消费者基础

在 Kafka 中，生产者负责把消息写入主题分区，消费者负责从分区拉取消息并推进偏移量。真正把这两端串起来的，是序列化、反序列化、偏移量提交和消费者组再平衡四个机制。本节以订单事件为例，用 Java 和 Python 各写一个最小可运行的生产者与消费者，并观察它们的行为。

一、序列化与反序列化：字节是唯一通行证

Kafka 的 broker 只存储字节数组，不关心消息内容。生产者发送前必须把对象转成字节，消费者收到后必须把字节还原成对象。这一对操作就是序列化与反序列化。

以订单事件为例，我们定义一个简单结构：订单号、用户号、金额、时间戳。Java 中常用 StringSerializer 处理键，用 JsonSerializer 或自定义序列化器处理值；Python 中常用 json.dumps 转成 UTF-8 字节。关键约定是：生产者的序列化格式必须与消费者的反序列化格式严格一致，否则会出现解析异常或乱码。

// Java 生产者核心配置
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("enable.idempotence", "true");

ProducerRecord<String, String> record =
    new ProducerRecord<>("orders", orderId, json);
producer.send(record, (meta, ex) -> {
    if (ex == null) System.out.println("offset=" + meta.offset());
    else ex.printStackTrace();
});

Python 版本更简洁，但语义相同：

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    key_serializer=lambda k: k.encode("utf-8"),
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    acks="all",
)
producer.send("orders", key="O1001", value={"user": "U7", "amount": 99.5})
producer.flush()

注意 acks=all 与幂等性配置：前者要求所有同步副本确认，后者避免重试导致重复写入。这两项是生产端可靠性的基础。

二、偏移量提交：消费者如何记住读到哪里

消费者读取消息后，需要把“已处理到哪个偏移量”记录到 Kafka 内部主题 __consumer_offsets 中，这就是偏移量提交。提交方式有两种：

自动提交：enable.auto.commit=true，默认每 5 秒提交一次。实现简单，但可能在消息尚未处理完时就提交，导致消息丢失。
手动提交：关闭自动提交，在处理完成后调用 commitSync() 或 commitAsync()。前者阻塞但可靠，后者异步但可能失败。

一个常见的正确模式是“先处理、后提交”：

// Java 手动提交
props.put("enable.auto.commit", "false");
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        process(r.value());          // 先处理业务
    }
    consumer.commitSync();           // 再提交偏移量
}

Python 中对应 enable_auto_commit=False，处理完后调用 consumer.commit()。判断是否提交成功，可以观察 commit() 是否抛出异常，或通过 consumer.committed(tp) 查询当前已提交偏移量。

如果业务处理本身是幂等的，自动提交也可以接受；如果业务涉及扣款、发券等不可重复操作，必须使用手动提交并配合幂等设计。

三、消费者组与再平衡：分区如何重新分配

多个消费者可以组成一个消费者组，共同消费一个主题。Kafka 保证同一分区在同一时刻只被组内一个消费者消费。当消费者加入、离开或崩溃时，组协调器会触发再平衡，重新分配分区。

再平衡的触发条件包括：消费者主动离开、心跳超时、订阅主题分区数变化、以及新消费者加入。再平衡期间，整个组会短暂停止消费，因此频繁再平衡会显著影响吞吐。

下面这张图展示了消费者组再平衡前后的分区分配变化，以及偏移量提交在其中的位置：

四、Java 与 Python 消费者实现要点

Java 消费者使用 KafkaConsumer，核心循环是 poll。需要设置 group.id、auto.offset.reset（无提交偏移量时从 earliest 还是 latest 开始）、max.poll.records 等参数。Python 使用 KafkaConsumer，参数名略有不同，但语义一致。

# Python 消费者
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "orders",
    bootstrap_servers="localhost:9092",
    group_id="order-group",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    value_deserializer=lambda b: json.loads(b.decode("utf-8")),
)
for msg in consumer:
    print(msg.partition, msg.offset, msg.value)
    consumer.commit()

观察再平衡最直接的方式是查看消费者日志中的 Rebalance 或 Assigning partitions 信息，也可以通过 kafka-consumer-groups.sh --describe --group order-group 查看每个分区的当前偏移量、日志末端偏移量和滞后量（LAG）。LAG 持续增大说明消费速度跟不上生产速度。

五、参数与行为对照

参数/行为	Java	Python	影响
序列化	key.serializer / value.serializer	key_serializer / value_serializer	决定字节格式，两端必须匹配
自动提交	enable.auto.commit	enable_auto_commit	true 简单但可能丢消息
提交方式	commitSync / commitAsync	commit()	同步可靠，异步高效
起始位置	auto.offset.reset	auto_offset_reset	earliest 从头，latest 从新
再平衡	session.timeout.ms / heartbeat.interval.ms	session_timeout_ms / heartbeat_interval_ms	超时越短越易触发再平衡

再平衡不是异常，而是消费者组的正常协调机制。但如果日志中频繁出现再平衡，通常意味着消费者处理时间过长、心跳超时或网络不稳定，需要检查 max.poll.interval.ms 与单批处理耗时。

把以上四点串起来：生产者用序列化把对象变成字节写入分区；消费者用反序列化还原对象；处理完成后提交偏移量；当组内成员变化时，再平衡重新分配分区，消费者从已提交偏移量继续读取。理解这条链路，就掌握了 Kafka 批流管道两端的基本工作方式。