4、Kafka生产者详解：发送流程、分区策略、ACK机制、重试机制

好，咱们今天来啃一块硬骨头——Kafka生产者。说实话，很多人在用Kafka的时候，关注点都在消费者那边，觉得生产者就是往里面塞数据，没什么好研究的。但我做了这么多年实时数仓，踩过的坑有一半都出在生产者身上。你想想看，数据都还没进Kafka呢，后面再好的架构也白搭。

所以这一章，我会把生产者的四个核心机制掰开揉碎了讲：发送流程、分区策略、ACK机制、重试机制。每个点我都会结合项目里的真实案例，告诉你哪些地方容易翻车。

4.1 生产者的整体发送流程

先看一张整体流程图，把生产者的内部脉络理清楚。

整个流程其实不复杂，我拆成几个阶段来讲。

4.2 发送流程详解

生产者的发送流程，说白了就是一条流水线。数据从业务代码里出来，经过几道工序，最后通过网络发到Broker。我习惯把它分成两个阶段：主线程处理和Sender线程发送。

4.2.1 主线程处理阶段

这个阶段发生在你的业务线程里，一共四步：

1. 拦截器链：你可以配置多个拦截器，按顺序执行。我一般在拦截器里做两件事：统一给消息加时间戳、做简单的数据脱敏。
2. 序列化器：把Key和Value从对象转成字节数组。Kafka自带的StringSerializer、ByteArraySerializer够用，但如果你用Avro或Protobuf，记得自己实现序列化器。
3. 分区器：决定这条消息去哪个分区。默认分区器逻辑我下面会细讲。
4. 消息累加器：这是最核心的一步。消息不会立刻发出去，而是先放进RecordAccumulator。每个分区对应一个双端队列（Deque），队列里存的是批次（Batch）。

4.2.2 Sender线程发送阶段

Sender线程是后台独立运行的，它干三件事：

• 从RecordAccumulator里拉取已经准备好的批次（批次满了或者linger.ms到了）
• 把属于同一个Broker的请求合并成一个Request，减少网络连接数
• 通过Selector发送网络请求，并处理响应

这里有个细节很多人不知道：Sender线程拉取批次的时候，不是一次拉一个，而是把同一个Broker上的所有分区的批次打包成一个请求。这样做的好处是减少网络开销，坏处是如果其中一个分区出问题，整个请求都得重试。

4.3 分区策略

分区策略决定了你的数据怎么分布。我遇到过最惨的一次事故，就是因为分区策略没配好，导致某个分区数据量是其他分区的100倍，直接把这个分区的磁盘打满了。

4.3.1 默认分区策略

Kafka默认的分区器是DefaultPartitioner，它的逻辑很简单：

条件	行为
Key为null	轮询（Round-Robin）分配到所有可用分区
Key不为null	对Key的哈希值取模，分配到固定分区

这里有个坑：当Key为null时，Kafka 2.4版本之前用的是轮询，2.4版本之后改成了黏性分区（Sticky Partitioner）。黏性分区会尽量把一批消息发到同一个分区，等批次满了再换下一个。这样做的好处是减少网络请求次数，提升吞吐量。

4.3.2 自定义分区策略

当默认策略满足不了你的时候，就该自己写分区器了。我举个例子：假设你要按业务类型分区，把订单数据发到分区0-2，支付数据发到分区3-5。

public class BusinessPartitioner implements Partitioner {
    private static final Map<String, Integer> BUSINESS_MAP = new HashMap<>();
    static {
        BUSINESS_MAP.put("order", 0);
        BUSINESS_MAP.put("payment", 3);
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String businessType = extractBusinessType(value);
        Integer basePartition = BUSINESS_MAP.get(businessType);
        if (basePartition == null) {
            // 未知业务类型，用默认分区器
            return DefaultPartitioner.INSTANCE.partition(
                topic, key, keyBytes, value, valueBytes, cluster);
        }
        // 在basePartition开始的连续分区中轮询
        int partitionCount = cluster.partitionCountForTopic(topic);
        return basePartition + (int)(Math.random() * 3) % 3;
    }

    @Override
    public void close() {}

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

写自定义分区器的时候，我建议你注意两点：一是要处理分区数变化的情况，二是要保证分区逻辑的确定性，不然同样的数据每次发到不同分区，消费端就乱套了。

4.4 ACK机制

ACK机制决定了生产者的可靠性级别。说白了就是：你发出去的消息，Broker要给你一个什么样的确认，你才放心？

4.4.1 三种ACK级别

ACK值	含义	可靠性	延迟	适用场景
acks=0	不等待任何确认	最低	最低	日志、监控等可丢失数据
acks=1	Leader分区写入成功即确认	中等	中等	大部分业务场景（默认值）
acks=-1（all）	所有ISR副本写入成功才确认	最高	最高	金融、交易等强一致性场景

我个人的建议是：除非你能接受丢数据，否则别用acks=0。我在一个日志采集项目里见过有人用acks=0，结果Broker挂了，丢了半小时的数据，排查了半天才发现是ACK级别的问题。

4.4.2 ACK=all的注意事项

用acks=all的时候，有个参数必须关注：min.insync.replicas。这个参数指定了ISR（In-Sync Replicas）中至少要有多少个副本同步成功，才算写入成功。

举个例子：你的Topic副本数是3，min.insync.replicas设成2。那么当acks=all时，只要Leader和任意一个Follower写入成功，就算成功。但如果只有一个副本在线，生产者就会收到异常。

4.5 重试机制

重试机制是生产者的最后一道防线。网络抖动、Leader选举、磁盘满……这些情况都会导致发送失败。没有重试机制，你的数据就丢了。

4.5.1 重试参数配置

参数	默认值	说明
retries	Integer.MAX_VALUE	重试次数（2.0版本后默认无限重试）
retry.backoff.ms	100	每次重试的间隔时间
delivery.timeout.ms	120000	发送超时时间（包含重试时间）
request.timeout.ms	30000	单次请求的超时时间

这里有个容易混淆的地方：retries和delivery.timeout.ms的关系。Kafka 2.0之后，retries默认是无限大，但delivery.timeout.ms限制了总时间。也就是说，重试不是无限进行的，超过delivery.timeout.ms还没成功，就放弃了。

4.5.2 幂等性生产者

重试会带来一个问题：消息重复。比如你发了一条消息，Broker写成功了，但ACK在网络中丢了，生产者以为失败了就重试，结果Broker又写了一遍。消费端就会收到两条相同的数据。

解决办法是开启幂等性：

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

开启后，生产者会给每条消息加一个序列号（Producer ID + 序列号），Broker会去重。同一个序列号的消息只会写入一次。

4.5.3 事务性生产者

如果你需要跨分区、跨Topic的原子性写入，那就得上事务了。事务性生产者保证：要么所有消息都写入成功，要么全部回滚。

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

事务性生产者我用的不多，因为性能开销比较大。只有在需要精确一次语义（Exactly-Once）的场景下，比如实时数仓的端到端一致性，我才会用。

4.6 避坑指南

最后，我把自己这些年踩过的坑总结一下：

• 不要盲目调大batch.size：批次太大，内存压力大，而且一条消息失败会导致整个批次重试。我一般从16KB开始，压测后逐步调大。
• 注意max.in.flight.requests.per.connection：默认是5。如果开启幂等性，这个值不能超过5。如果没开幂等性，设成1可以保证消息顺序，但吞吐量会下降。
• buffer.memory别设太小：默认32MB。如果发送速度快于Broker处理速度，缓冲区满了就会阻塞。我一般设成64MB或128MB。
• 重试要配合超时时间：别只设retries不设delivery.timeout.ms，否则网络分区时生产者会一直重试，把资源耗尽。

嗯，这一章的内容就这些。生产者的核心机制其实不复杂，但每个参数背后都有它的设计意图。理解了这些，你就能根据业务场景灵活配置了。