4. 生产者与消费者客户端开发

Kafka 客户端开发的核心，不是“能发能收”，而是“在异常、重试、并发和重启条件下仍然不丢不重”。本章围绕 Producer 与 Consumer 的关键配置、序列化、批量与压缩、重试与幂等、以及偏移量提交策略展开，目标是写出可上线、可观测、可恢复的生产消费程序。

一、Producer 关键配置与发送链路

一条消息从 send() 到落盘，会经历拦截器、序列化器、分区器、RecordAccumulator 缓冲、Sender 线程批量发送、Broker 写入这几个阶段。理解这条链路，才能理解各参数的作用位置。

最关键的几个参数：

• bootstrap.servers：仅用于获取集群元数据，不是全部 Broker 列表。
• key.serializer / value.serializer：序列化器全限定类名，必须与发送类型匹配。
• acks：0 不等确认（可能丢）、1 Leader 落盘即返回、all（或 -1）ISR 全部确认，配合幂等必须为 all。
• retries 与 retry.backoff.ms：可重试异常的重试次数与退避。
• linger.ms 与 batch.size：批量发送的两个开关，前者按时间攒批，后者按字节攒批。
• compression.type：none、gzip、snappy、lz4、zstd。
• enable.idempotence=true：开启幂等，配合 acks=all、retries>0、max.in.flight.requests.per.connection<=5。

下面这张图展示了 Producer 的发送链路与各参数的作用点，以及幂等性依赖的配置约束。

二、序列化机制

Kafka 只认字节数组，序列化器负责对象到字节的转换。内置的有 StringSerializer、ByteArraySerializer、IntegerSerializer 等。生产环境更常用 Avro、Protobuf、JSON Schema，配合 Schema Registry 做模式演进与兼容性校验。

自定义序列化器需实现 Serializer<T>，重写 serialize(topic, data) 与 configure()。注意：序列化器必须线程安全，因为 Producer 是线程安全的，多个线程可能共享同一个序列化器实例。反序列化器同理，且必须与序列化器严格配对，否则会出现 SerializationException 或乱码。

三、批量发送与压缩

批量发送的两个参数：batch.size 默认 16KB，linger.ms 默认 0。当 linger.ms=0 时，Sender 线程只要缓冲区有数据就立即发送，吞吐低但延迟最低。把 linger.ms 调到 5～50ms，可以让更多消息攒进同一批次，显著提升吞吐，代价是增加延迟。

压缩在批次级别进行，压缩的是整个 RecordBatch，因此批量越大，压缩收益越高。选择建议：

• lz4：吞吐与压缩比均衡，延迟低，通用首选。
• zstd：压缩比最高，CPU 消耗略高，适合带宽紧张场景。
• gzip：压缩比高但 CPU 重、延迟高，逐渐被 zstd 替代。
• snappy：老牌选择，压缩比一般。

压缩由 Producer 完成，Broker 原样存储，Consumer 解压。因此压缩类型写在消息批次头里，消费端无需额外配置即可解压。

四、重试与幂等性

可重试异常包括 NotLeaderForPartitionException、LeaderNotAvailableException、NetworkException、TimeoutException 等；不可重试异常如 RecordTooLargeException、SerializationException，重试无意义。

仅靠重试会带来重复：Broker 已写入但 ACK 丢失，Producer 重试就会写第二条。幂等性通过 Producer ID（PID）+ 分区级 Sequence Number 解决：Broker 对同一 PID 的同一分区维护序列号，重复批次直接丢弃。注意幂等只保证单分区、单会话内不重，跨会话（Producer 重启后 PID 变化）不保证，跨分区也不保证。

若需要跨会话、跨分区的精确一次，需要事务：设置 transactional.id，调用 initTransactions()、beginTransaction()、commitTransaction()，并配合 Consumer 的 isolation.level=read_committed。

五、Consumer 配置与偏移量提交

Consumer 的核心参数：

• group.id：消费者组标识，同组内分区互斥分配。
• enable.auto.commit：默认 true，自动提交，容易丢或重。
• auto.commit.interval.ms：自动提交间隔，默认 5000ms。
• auto.offset.reset：无提交偏移时的起点，earliest / latest / none。
• max.poll.records：单次 poll 返回的最大记录数，默认 500。
• max.poll.interval.ms：两次 poll 的最大间隔，默认 5 分钟，超时触发 rebalance。
• isolation.level：read_uncommitted（默认）或 read_committed。

偏移量提交有两种方式：自动提交和手动提交。手动提交又分同步 commitSync() 和异步 commitAsync()。推荐模式是“处理完再提交”，即先处理消息，再提交偏移，保证至少一次语义。

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("group.id", "order-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");
props.put("max.poll.records", "200");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        process(r);   // 业务处理
    }
    if (!records.isEmpty()) {
        consumer.commitSync();  // 处理完再提交
    }
}

如果处理逻辑可能失败，应把偏移量与业务结果放在同一事务里（如数据库事务），或使用“处理成功才提交”的模式。异步提交 commitAsync() 吞吐更高，但失败不会自动重试，通常在 onComplete 回调里记录失败，并在关闭前用 commitSync() 兜底。

六、可靠生产消费的检查清单

环节	关键配置	目标
Producer 可靠性	acks=all, retries>0, enable.idempotence=true	不丢不重（单分区单会话）
Producer 吞吐	linger.ms=10~50, batch.size=64KB~256KB, compression.type=lz4/zstd	提升吞吐、降低带宽
Consumer 语义	enable.auto.commit=false, 处理完 commitSync()	至少一次
Consumer 稳定性	max.poll.records 调小, max.poll.interval.ms 合理	避免 rebalance 风暴
精确一次	transactional.id + isolation.level=read_committed	跨分区跨会话 EOS

最后强调：所有可靠性配置都要配合监控。关注 Producer 的 record-error-rate、record-retry-rate、batch-size-avg，Consumer 的 records-lag-max、commit-latency-avg、rebalance-rate。只有指标可见，配置才真正可控。