Kafka 生产者详解：发送流程、分区策略、ACK 机制、重试机制、自定义拦截器

各位同学，今天我们来啃一块硬骨头——Kafka 生产者。说实话，很多人在用 Kafka 的时候，只关心怎么发消息，却忽略了生产者的内部机制。结果线上出问题的时候，一脸懵。我当年刚接触 Kafka 时也踩过不少坑，今天就把这些经验掰开揉碎了讲给你听。

1. 生产者的核心发送流程

Kafka 生产者发送消息，不是直接扔给 Broker 的。它内部有一套完整的流水线。我习惯把它分成五个步骤：

1. 消息封装：把业务数据包装成 ProducerRecord 对象，指定 topic、value，可选 key 和 partition。
2. 序列化：把对象转成字节数组。Kafka 内置了 String、Integer 等序列化器，你也可以自定义。
3. 分区器：决定这条消息发到哪个分区。如果指定了 partition，直接使用；否则根据 key 计算。
4. 缓冲区：消息进入 RecordAccumulator，这是一个内存队列。默认 32MB，每个批次默认 16KB。
5. Sender 线程：后台线程不断从缓冲区拉取批次，打包成请求，发送给 Broker。

嗯，这里要注意：缓冲区的大小和批次大小，直接影响你的发送性能。我在项目中遇到过，有人把 buffer.memory 设得太小，结果高并发时频繁触发阻塞，TPS 直接腰斩。

2. 分区策略：消息到底去哪了？

分区策略决定了消息落在哪个分区。Kafka 默认提供了几种策略，你也可以自己写。我常用的有三种：

策略	原理	适用场景
轮询策略	消息均匀分配到所有分区	无 key 消息，追求负载均衡
哈希策略	对 key 的 hash 值取模，相同 key 进入同一分区	需要保证消息有序性（按 key）
自定义策略	实现 Partitioner 接口，自己写逻辑	特殊业务需求，比如按地域路由

你想想看，如果分区数不够，或者分区策略不合理，再好的集群也扛不住。所以设计分区时，我建议先估算峰值 TPS，再决定分区数。一般经验是：分区数 = 目标吞吐量 / 单个分区吞吐量，再乘以 1.5 的冗余系数。

3. ACK 机制：消息到底发成功没？

ACK 是 Kafka 保证消息可靠性的核心参数。它有三个取值：

ACK 值	含义	可靠性	性能
acks=0	生产者发完就认为成功，不等待任何确认	最低	最高
acks=1	Leader 分区写入成功即返回确认	中等	中等
acks=all	所有 ISR 副本都写入成功才返回确认	最高	最低

说白了，ACK 机制就是在性能和可靠性之间做 trade-off。你选 0，就是裸奔；选 all，就是穿防弹衣。看你的业务能承受多大风险。

4. 重试机制：失败了怎么办？

网络抖动、Broker 宕机，这些在生产环境太常见了。Kafka 生产者内置了重试机制，核心参数有两个：

• retries：重试次数，默认 Integer.MAX_VALUE（2.0 版本后）。我建议设成 3-5 次，别无限重试。
• retry.backoff.ms：重试间隔，默认 100ms。我习惯设成 200-500ms，给 Broker 一点恢复时间。

嗯，这里有个坑：重试可能导致消息乱序。比如你发了消息 A 和 B，A 失败了在重试，B 成功了，结果 A 重试成功后反而排在 B 后面。怎么解决？

我记得有一次，线上某个 topic 的写入量突然暴增，导致 Broker 负载过高，大量请求超时。重试机制疯狂触发，结果消息堆积在缓冲区，最后 OOM 了。从那以后，我养成了一个习惯：重试次数不要设太大，同时监控缓冲区使用率。

5. 自定义拦截器：在发送前后插一脚

拦截器是 Kafka 提供的一个扩展点。你可以实现 ProducerInterceptor 接口，在消息发送前或发送后执行自定义逻辑。比如：

• 统计发送成功率
• 给消息打上时间戳
• 过滤掉某些敏感数据
• 记录审计日志

来看一个简单的拦截器示例：

public class MyInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 在消息头部添加发送时间
        record.headers().add("sendTime", 
            String.valueOf(System.currentTimeMillis()).getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // 发送完成后，记录成功或失败
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
        }
    }

    @Override
    public void close() {
        // 清理资源
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 初始化配置
    }
}

然后在生产者配置里加上：

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
    "com.example.MyInterceptor");