4、Kafka生产者详解：从客户端开发到消息可靠投递

好，咱们进入生产者的世界。说实话，Kafka 生产者这块内容，是我在实际项目中踩坑最多的地方。很多初学者觉得生产者就是发个消息嘛，有什么难的？但等你上了生产环境，丢消息、重复消费、分区倾斜这些问题会一个个找上门来。

这一章，我会把生产者从客户端开发到消息可靠投递的完整链路讲清楚。你跟着走一遍，基本就能应付大部分生产场景了。

4.1 生产者客户端开发：从零写一个生产者

先来个最基础的生产者代码。我习惯用 Java 客户端，但原理同样适用于其他语言。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

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

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "hello kafka");
producer.send(record);
producer.close();

这段代码看着简单，但有几个关键点你得注意：

• bootstrap.servers：别只配一个 broker 地址，配两三个。万一那个挂了，客户端还能连别的。
• 序列化器：Kafka 只存字节数组，你得告诉它怎么把对象转成字节。StringSerializer 是最常用的，但如果你传的是对象，记得用 Avro 或 Protobuf。
• ProducerRecord：消息载体，包含 topic、partition（可选）、key（可选）、value。

4.2 消息发送方式：同步 vs 异步

Kafka 的 send() 方法其实是异步的。它返回一个 Future 对象。你可以选择等它完成，也可以不管它。

4.2.1 同步发送

try {
    RecordMetadata metadata = producer.send(record).get();
    System.out.println("消息发送成功，offset: " + metadata.offset());
} catch (Exception e) {
    e.printStackTrace();
}

调用了 .get()，当前线程就会阻塞，直到 broker 返回确认。这种方式简单，但吞吐量低。我一般只在测试或关键业务（比如订单创建）里用。

4.2.2 异步发送 + 回调

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
        } else {
            System.out.println("发送成功，offset: " + metadata.offset());
        }
    }
});

这才是生产环境的主流用法。不阻塞主线程，失败时通过回调处理。我曾经在一个日志采集系统里，用异步发送把吞吐量从每秒 2000 条提到了 15000 条。

4.3 消息分区策略：数据该去哪？

Kafka 的分区机制，说白了就是决定一条消息落到哪个分区里。分区策略直接影响数据分布和消费顺序。

4.3.1 默认分区器

• 有 key：对 key 的 hash 值取模，相同 key 的消息永远进同一个分区。保证顺序。
• 无 key：轮询（Round-Robin）或粘性分区（Sticky Partitioner）。粘性分区是 2.4 版本引入的，它会先攒一批消息发到同一个分区，减少网络开销。

4.3.2 自定义分区策略

有时候默认策略不满足需求。比如你想把某个业务线的消息全打到分区 0。那就自己实现 Partitioner 接口：

public class BusinessPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        if ("order".equals(key)) {
            return 0; // 订单消息全进分区0
        }
        return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);
    }
}

然后在配置里指定：

props.put("partitioner.class", "com.example.BusinessPartitioner");

4.4 消息可靠性保证：ACKS 参数

ACKS 是生产者最重要的可靠性参数。它决定了生产者要等多少个副本确认，才算发送成功。

我个人的建议：

• 日志、监控、非关键数据：用 acks=1，性能好，偶尔丢几条能接受。
• 订单、支付、交易：用 acks=all，配合 min.insync.replicas=2，保证至少两个副本写入才算成功。
• acks=0：除非你明确知道自己在做什么（比如丢数据无所谓），否则别用。

4.5 幂等性与事务：解决重复和一致性问题

4.5.1 幂等性生产者

幂等性解决的是「消息重复」问题。你想想看，如果网络抖动，生产者重试发送，broker 可能收到两条一模一样的消息。幂等性就是让 broker 能识别并去重。

开启方式很简单：

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

开启后，producer 会为每条消息分配一个唯一的序列号（Producer ID + 序列号）。broker 根据序列号去重。注意，开启幂等性要求 acks=all 且 retries > 0，否则会报错。

4.5.2 事务性生产者

事务解决的是「原子性」问题。比如你从一个 topic 消费，处理后写入另一个 topic，这两个操作要么都成功，要么都失败。

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

事务的底层原理是引入了一个 Transaction Coordinator，它负责记录事务状态。开启事务需要配置：

props.put("transactional.id", "my-transactional-id");

注意，transactional.id 必须是唯一的，producer 重启后要能恢复之前的事务状态。

4.6 知识体系总览

下面这张图，把生产者核心链路串起来了。你可以对照着回顾一下：

从客户端到 broker，每一步都有对应的配置和策略。你不需要一次性全记住，但遇到问题时，能快速定位到是哪个环节出了问题，这就够了。