📚 课程目录第4章生产者原理与API实践

4. 生产者原理与API实践：发送流程、分区策略、ACK机制、幂等性与事务

这一章，我们来啃生产者这块硬骨头。说实话，Kafka 生产者的坑比消费者多得多。我见过太多线上事故，都是因为对发送流程理解不透彻导致的。咱们一步步拆开看。

4.1 生产者的核心发送流程

先看整体架构，心里有个地图。

一句话总结：生产者不是一条消息直接发到 Broker，而是先经过拦截器、序列化器、分区器，攒成批次再发送。

流程其实不复杂。你创建一个 ProducerRecord，指定 topic、value，可选 key 和 partition。然后它依次经过拦截器、序列化器、分区器，最后进入 RecordAccumulator。Sender 线程从 Accumulator 拉取批次，发到 Broker。

这里有个细节我吃过亏——拦截器执行顺序。如果你配了多个拦截器，它们按配置顺序组成链。我曾经在拦截器里做消息加密，结果后面又加了个压缩拦截器，顺序搞反了，解密端直接崩溃。嗯，拦截器顺序一定要想清楚。

4.2 分区策略：数据该去哪？

分区策略决定了消息落到哪个分区。Kafka 默认提供了几种策略，你也可以自己写。

策略	说明	适用场景
RoundRobinPartitioner	轮询分配，消息均匀分布	无 key 消息，负载均衡
UniformStickyPartitioner	粘性分区，先攒一批再轮询	默认策略，吞吐更高
HashPartitioner	对 key 哈希取模	需要相同 key 进入同一分区
自定义分区器	实现 Partitioner 接口	业务定制路由

我的建议：如果你需要保证相同 key 的消息有序，用 HashPartitioner。如果只是单纯发日志，用默认的 UniformStickyPartitioner 就行，吞吐量更高。

自定义分区器怎么写？看代码：

public class OrderIdPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 假设 key 是订单ID，取最后一位决定分区
        String orderId = (String) key;
        int lastDigit = Integer.parseInt(orderId.substring(orderId.length() - 1));
        return lastDigit % cluster.partitionCountForTopic(topic);
    }

    @Override
    public void close() {}

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

配置时加上 partitioner.class=com.xxx.OrderIdPartitioner 就行。我在项目中遇到过一种情况：订单数据需要按地域路由，但地域信息在 value 里。这时候自定义分区器就派上用场了——从 value 里解析出地域字段，再决定分区。

4.3 ACK机制：到底等不等确认？

ACK 是生产者的可靠性开关。Kafka 提供了三个级别：

acks 值	行为	可靠性	延迟
0	发完就走，不等待确认	最低	最低
1	等待 leader 确认	中等	中等
all (-1)	等待所有 ISR 副本确认	最高	最高

我曾经踩过的坑：有一次做日志采集，为了追求吞吐把 acks 设成 0。结果 Broker 挂了，日志全丢。后来改成 acks=1，配合重试机制，才稳住。如果你对数据丢失零容忍，必须用 acks=all。

但 acks=all 也不是万能的。它依赖 min.insync.replicas 参数。比如你设置副本数为 3，min.insync.replicas=2，那么至少 2 个副本确认才算成功。如果只剩 1 个副本在线，生产者会收到 NotEnoughReplicasException。

4.4 幂等性：消息不会重复

幂等性解决什么问题？网络重试导致的消息重复。你想想看，生产者发送消息后没收到 ACK，重试了一次，结果 Broker 收到了两条一模一样的消息。

Kafka 的幂等性方案很巧妙：

每个生产者有一个唯一的 producerId（PID）
每条消息带一个单调递增的 sequence number
Broker 按 <PID, 分区> 去重，重复的 seq 直接丢弃

开启方式很简单：

props.put("enable.idempotence", true);

注意，开启幂等性后，acks 会自动变成 all，retries 会变成 Integer.MAX_VALUE。这是 Kafka 的强制绑定——幂等性必须配合最高可靠性。

核心要点：幂等性只保证单分区内的精确一次语义。跨分区的事务，需要用到事务 API。

4.5 事务：跨分区原子写入

事务是幂等性的升级版。它保证一批消息要么全部写入成功，要么全部回滚。常用于「读取-处理-写入」的 ETL 场景。

事务的流程分三步：

初始化事务：producer.initTransactions()
开启事务：producer.beginTransaction()
提交或回滚：producer.commitTransaction() / producer.abortTransaction()

代码示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "txn-order-001");  // 必须设置
props.put("enable.idempotence", true);

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

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "order-1", "data1"));
    producer.send(new ProducerRecord<>("payments", "pay-1", "data2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

注意：transactional.id 必须设置，且每个生产者实例要唯一。如果两个生产者用同一个 transactional.id，Kafka 会认为前一个挂了，强制中止其事务。我见过有人部署多实例时忘了改这个 ID，结果事务互相打断，数据乱成一锅粥。

事务的底层依赖一个特殊的内部 topic——__transaction_state。Broker 上的 Transaction Coordinator 负责管理事务状态。每次提交事务，会写入一个事务标记（commit marker），消费者读取时根据标记决定是否可见。

4.6 生产者的关键参数调优

最后，我列几个我常用的参数配置，供你参考：

参数	默认值	我的建议
`batch.size`	16KB	32KB~64KB，吞吐更高
`linger.ms`	0	5~20ms，攒批减少请求数
`compression.type`	none	snappy 或 lz4，CPU换带宽
`max.in.flight.requests.per.connection`	5	开启幂等性后建议设为 1 或 5
`retries`	0	开启幂等性后自动设为 MAX

个人习惯：我一般先设 batch.size=32768、linger.ms=10，然后压测看吞吐和延迟。如果延迟敏感，把 linger.ms 降到 0。如果吞吐优先，可以继续加大 batch.size 到 128KB。

好了，生产者这块的核心内容就这些。发送流程、分区策略、ACK 机制、幂等性、事务——每个点都值得你在项目中亲手试一遍。尤其是事务，别看代码简单，真正跑起来才会发现各种边界情况。