4. Kafka 生产者深入：发送流程、分区策略、acks 与 retries、幂等性与事务、自定义 Partitioner

本章深入剖析 Kafka 生产者的核心工作机制。理解发送流程、分区策略、可靠性参数、幂等性与事务，以及自定义分区器，是构建高吞吐、高可靠实时图谱数据管道的基础。我们将从一条消息的诞生开始，追踪它如何被序列化、分区、缓冲并最终发送到 Broker。

4.1 生产者发送流程：从 ProducerRecord 到 Broker 确认

Kafka 生产者的发送流程并非简单的“发送即忘”，而是一个包含序列化、分区、缓冲、批量发送与异步回调的复杂过程。下图展示了消息从创建到确认的完整生命周期。

4.2 分区策略：数据分布的核心逻辑

分区策略决定了消息被发送到 Topic 的哪个 Partition。合理的分区策略是实现数据局部性、负载均衡与有序性的关键。Kafka 提供了默认分区器，也支持用户自定义。

分区策略	触发条件	行为描述	适用场景
指定分区	ProducerRecord 显式指定 partition	直接使用指定的分区号，忽略 key 与分区器	严格数据路由，如按实体 ID 固定分区
Key 哈希	消息包含 key，且未指定分区	对 key 进行哈希（murmur2），对分区数取模	保证相同 key 进入同一分区，实现局部有序
轮询 (Round-Robin)	消息无 key，且未指定分区	依次将消息分配到各个分区，负载均衡	日志、监控等无需有序性的场景
粘性分区 (Sticky)	消息无 key（Kafka ≥ 2.4）	先填满一个批次再切换分区，减少网络开销	高吞吐无 key 场景，性能优于轮询

4.3 acks 与 retries：可靠性 vs 吞吐量的权衡

acks 参数决定了生产者等待 Broker 确认的级别，直接影响数据可靠性。结合 retries 和 max.in.flight.requests.per.connection，可以精细控制发送的容错行为。

enable.idempotence=true 使用，无限重试直到成功 • max.in.flight.requests.per.connection=1：防止重试导致消息乱序（未启用幂等性时） • max.in.flight.requests.per.connection=5：启用幂等性后可设置为 5，提升吞吐 • delivery.timeout.ms：控制发送超时（默认 120s），超过后放弃重试

4.4 幂等性与事务：精确一次语义

幂等性（Idempotence）和事务（Transaction）是 Kafka 实现精确一次语义（EOS）的两大基石。幂等性防止单分区内的重复消息，事务则保证跨分区跨会话的原子写入。

4.4.1 幂等性生产者

启用 enable.idempotence=true 后，生产者会为每个批次分配一个 Producer ID (PID) 和 序列号 (Sequence Number)。Broker 会检查序列号，如果收到重复序列号则丢弃该批次，从而避免网络重试导致的重复写入。

// 幂等性生产者配置示例
Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 启用幂等性后，以下参数会自动调整：
// acks = all
// retries = Integer.MAX_VALUE
// max.in.flight.requests.per.connection = 5
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

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

4.4.2 事务性生产者

事务性生产者在幂等性基础上，引入 Transaction ID 和 事务协调器，实现跨分区、跨会话的原子写入。典型应用场景是“读取-处理-写入”的流处理任务，如 Kafka Streams 或 Flink 的 Exactly-Once 保证。

特性	幂等性 (Idempotence)	事务 (Transaction)
作用范围	单分区内	跨分区、跨 Topic
重复消息	防止重复	防止重复
原子性	不支持	支持（原子提交/回滚）
跨会话恢复	不支持（PID 重启后变化）	支持（Transaction ID 持久化）
配置参数	enable.idempotence=true	transactional.id + enable.idempotence=true
性能开销	低	中（增加事务协调器通信）

// 事务性生产者配置示例
Properties props = new Properties();
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id-1");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// ... 序列化配置

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();  // 初始化事务

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic-a", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic-b", "key2", "value2"));
    producer.commitTransaction();  // 原子提交
} catch (Exception e) {
    producer.abortTransaction();   // 回滚
}

4.5 自定义 Partitioner：实现灵活的数据路由

当默认分区策略无法满足业务需求时，可以实现 org.apache.kafka.clients.producer.Partitioner 接口，自定义分区逻辑。例如，在实时图谱场景中，可能需要根据实体类型、地域或数据量进行分区。

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class EntityTypePartitioner implements Partitioner {

    private static final int USER_PARTITION = 0;
    private static final int DEVICE_PARTITION = 1;
    private static final int DEFAULT_PARTITION = 2;

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 根据 key 的前缀决定分区
        if (key == null) {
            return DEFAULT_PARTITION;
        }
        String keyStr = (String) key;
        if (keyStr.startsWith("user_")) {
            return USER_PARTITION;
        } else if (keyStr.startsWith("device_")) {
            return DEVICE_PARTITION;
        } else {
            // 其他 key 使用哈希取模
            return Math.abs(keyStr.hashCode()) % cluster.partitionCountForTopic(topic);
        }
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

在生产者配置中指定自定义分区器：

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
          "com.example.EntityTypePartitioner");