4. Kafka 生产者深入：发送模式、分区策略、消息确认机制与幂等性

在实时特征管道中，生产者是数据进入 Kafka 的入口。本章深入剖析生产者的核心行为：发送模式如何影响吞吐与延迟，分区策略如何决定数据分布，acks 参数如何平衡可靠性与性能，以及重试与幂等性如何保证端到端的一致性。理解这些机制是构建稳定、高性能特征管道的基础。

4.1 发送模式：同步与异步

Kafka 生产者客户端提供两种发送模式，本质区别在于 send() 方法是否阻塞等待 Broker 确认。

4.2 分区策略

分区策略决定了消息被写入哪个 Partition。合理的分区策略能保证数据均匀分布、维持有序性并支持局部性优化。

4.2.1 默认分区器逻辑

• Key 非空： 对 Key 进行哈希（murmur2），然后对分区数取模，相同 Key 始终进入同一分区。
• Key 为空： 使用粘性分区策略（Sticky Partitioner），将一批消息发送到同一个分区以减少请求数量，提升吞吐。

4.2.2 自定义分区策略

实现 org.apache.kafka.clients.producer.Partitioner 接口，重写 partition() 方法。常见场景包括：

• 基于业务 ID 路由： 将同一用户/设备的特征数据路由到同一分区，保证局部有序。
• 基于数据量均衡： 轮询或按分区负载动态分配，避免数据倾斜。
• 基于地理位置： 将数据路由到最近的 Broker 所在分区。

4.3 消息确认机制（acks）

acks 参数控制生产者需要收到多少个副本的确认才算发送成功。这是可靠性与延迟之间的核心权衡。

4.4 重试机制与幂等性

4.4.1 重试机制

当生产者发送失败（如网络抖动、Leader 选举），Kafka 客户端会自动重试。关键参数：

• retries：重试次数，默认 Integer.MAX_VALUE（2.1+ 版本）。
• retry.backoff.ms：重试间隔，默认 100ms。
• delivery.timeout.ms：发送超时上限，默认 120s，超过此时间不再重试。

4.4.2 幂等性

幂等性保证生产者发送的每条消息在 Broker 端只被持久化一次，即使重试也不会产生重复。启用方式：

// 生产者配置
props.put("enable.idempotence", true);
// 当幂等性启用时，以下参数会被自动调整：
// acks = all
// max.in.flight.requests.per.connection = 5（允许乱序但去重）
// retries = Integer.MAX_VALUE

幂等性原理：生产者每次初始化会分配一个 Producer ID (PID)，每条消息附带一个单调递增的 Sequence Number。Broker 端会缓存每个 Partition 最近 5 个 Sequence Number，如果收到重复序号则直接丢弃。

4.5 综合配置示例

以下是一个面向实时特征管道的高可靠生产者配置模板：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

// 发送模式：异步 + Callback
props.put("acks", "all");                          // 最高可靠性
props.put("enable.idempotence", true);             // 幂等性，防止重试重复
props.put("max.in.flight.requests.per.connection", 5); // 幂等性下允许乱序
props.put("retries", Integer.MAX_VALUE);           // 无限重试
props.put("delivery.timeout.ms", 120000);          // 总超时 2 分钟
props.put("retry.backoff.ms", 100);                // 重试间隔 100ms

// 吞吐优化
props.put("batch.size", 16384);                    // 16KB 批次
props.put("linger.ms", 10);                        // 等待 10ms 凑批
props.put("compression.type", "snappy");           // 压缩减少网络开销
props.put("buffer.memory", 33554432);              // 32MB 发送缓冲区

KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);