4. Kafka 生产者：发送流程、同步/异步发送、acks 机制、retries 重试、幂等性

各位同学，今天我们来聊聊 Kafka 生产者的核心机制。这部分内容，说白了就是「消息怎么从你的应用跑到 Broker 上的」。我见过不少新手，上来就写个 producer.send()，结果消息丢了都不知道。嗯，咱们今天就把这块彻底讲透。

4.1 生产者的整体发送流程

先看一张总图，你就能明白消息的「一生」了。

整个流程其实就六步：

1. 创建 ProducerRecord：指定 topic、key、value，可选 partition 或 timestamp。
2. 拦截器链：你可以在这里做消息预处理、打日志、改内容。
3. 序列化：key 和 value 转成字节数组。我见过有人用 StringSerializer 传 JSON，结果 JSON 里带了个特殊字符，Broker 直接报错——后来我养成了习惯，生产环境一律用 Avro 或 Protobuf。
4. 分区器：决定消息去哪个分区。默认按 key 哈希，key 为 null 则轮询。
5. RecordAccumulator：消息先攒在内存缓冲区里，每个分区一个双端队列。
6. Sender 线程：后台线程从缓冲区拉取一批消息，打包成请求发出去。

4.2 同步发送 vs 异步发送

很多初学者搞不清这两个概念。其实很简单——同步发送就是发一条等一条，异步发送就是发完就走，回调来了再处理。

4.2.1 同步发送

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
try {
    RecordMetadata metadata = producer.send(record).get();  // 阻塞等待
    System.out.println("发送成功，offset: " + metadata.offset());
} catch (Exception e) {
    e.printStackTrace();
}

注意，.get() 会一直等到 Broker 返回响应。我曾经在一个高吞吐场景下用了同步发送，结果 TPS 直接掉到几百——因为网络往返时间成了瓶颈。所以，同步发送只适合对一致性要求极高、吞吐量不敏感的场景，比如金融交易日志。

4.2.2 异步发送

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        System.err.println("发送失败: " + exception.getMessage());
        // 这里可以写重试逻辑或记录死信
    } else {
        System.out.println("发送成功，offset: " + metadata.offset());
    }
});

异步发送是生产环境的主流选择。你想想看，Kafka 设计出来就是为了扛百万级 TPS 的，如果每条消息都同步等，那还怎么玩？

4.3 acks 机制：消息可靠性的「三档开关」

acks 参数决定了生产者要等多少个副本确认，才算发送成功。说白了，就是「你要多确认才放心」。

我在项目中遇到过这么一件事：有一次线上流量突增，我们把 acks 从 all 改成了 1，结果 Leader 节点刚好在那会儿宕机了，几十万条消息就这么丢了。嗯，从那以后，核心业务我坚决用 acks=all，非核心日志类业务才用 acks=1。

4.4 retries 重试机制

网络抖动、Leader 选举、磁盘满……这些情况都可能让发送失败。retries 就是 Kafka 给你的「第二次机会」。

// 生产者配置示例
props.put(ProducerConfig.RETRIES_CONFIG, 3);                // 重试次数
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);     // 重试间隔（毫秒）
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); // 总超时时间

这里有个坑：retries 默认是 Integer.MAX_VALUE（在 2.1 版本之后）。为什么？因为 Kafka 团队认为「只要还能重试，就不要轻易放弃」。但如果你不设 delivery.timeout.ms，消息可能会在缓冲区里重试几个小时——我见过一个案例，重试队列积压了上百万条消息，直接把内存撑爆了。

4.5 幂等性：解决「重复消息」的终极方案

你想想看，如果 Broker 已经写入了消息，但响应在网络传输中丢了，生产者会重试——这就导致同一条消息被写入多次。幂等性就是来解决这个问题的。

4.5.1 原理

开启幂等性后，每个 Producer 会有一个唯一的 Producer ID（PID），每条消息带一个 序列号（Sequence Number）。Broker 会检查序列号，如果发现重复，直接丢弃。

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

就这么一行配置。但注意，开启幂等性后，acks 会被强制设为 all，retries 会被设为 Integer.MAX_VALUE。因为幂等性要求「必须确认写入成功，且不能丢消息」。

4.5.2 局限性

• 单会话内有效：如果 Producer 重启，PID 会变，之前的序列号状态就丢了。重启后第一条消息如果重试，还是可能重复。
• 跨分区不保证：幂等性只保证单个分区内不重复，跨分区的事务需要用到 Kafka 事务 API。

4.6 综合配置示例

最后，给出一份我常用的生产配置，供你参考：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092,broker3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 可靠性配置
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   // 开启幂等性
props.put(ProducerConfig.ACKS_CONFIG, "all");                 // 等待所有 ISR 确认
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);  // 无限重试（配合超时时间）
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); // 总超时 2 分钟

// 吞吐量配置
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);          // 16KB 批次
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);               // 最多等待 5ms
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);    // 64MB 缓冲区

// 压缩配置
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy"); // 推荐 snappy，CPU 开销小

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

这套配置在我经历过的多个生产项目中都表现稳定。当然，具体参数要根据你的业务场景微调——比如日志类业务可以关掉幂等性、调小 acks，换取更高的吞吐量。