3. Kafka生产者API实战：创建生产者、发送消息的三种方式、分区策略与自定义分区器

好，咱们进入第三章。前面两章我们把Kafka的架构和核心概念捋了一遍，现在该动手了。这一章我带你手写Kafka生产者代码，把消息真正发到Broker里去。

说实话，生产者的API看起来简单，但坑不少。我在好几个项目里都见过同事因为没搞懂发送方式，导致数据丢了或者性能上不去。咱们今天就把这块彻底讲透。

3.1 创建生产者：三步走

创建一个Kafka生产者，说白了就三步：配参数、建对象、发消息。先看最基础的代码骨架：

Properties props = new Properties();
// 第一步：指定Broker地址
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);

嗯，这里有个细节我要强调一下。bootstrap.servers 不需要把所有Broker都写进去，写两三个就够了。为什么？因为客户端连上一个Broker之后，会通过元数据请求拿到整个集群的信息。我曾经见过有人把20个Broker全写进去，完全没必要。

3.2 发送消息的三种方式

消息发出去，到底要不要等结果？这就引出了三种发送方式。我按可靠性从低到高给你讲。

3.2.1 单向发送（Fire-and-Forget）

最简单粗暴的方式。消息扔出去就不管了，不管它到没到：

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

这种方式性能最高，但可靠性最低。适合什么场景？日志采集、监控指标这类丢了也无所谓的场景。我在做用户行为日志采集时用过，因为业务方说丢几条日志不影响统计结果。

3.2.2 同步发送

调用 send() 之后，调用 get() 阻塞等待结果：

try {
    ProducerRecord<String, String> record = 
        new ProducerRecord<>("my-topic", "key1", "value1");
    RecordMetadata metadata = producer.send(record).get();
    System.out.println("消息发送成功，偏移量：" + metadata.offset());
} catch (Exception e) {
    e.printStackTrace();
}

同步发送能拿到 RecordMetadata，里面有分区号、偏移量这些信息。但代价是性能差——每发一条消息都要等Broker确认。我一般只在核心交易链路里用，比如订单状态变更，丢一条可能赔钱的那种。

3.2.3 异步发送（带回调）

这是我最推荐的方式。发消息时不阻塞，但通过回调函数处理结果：

ProducerRecord<String, String> record = 
    new ProducerRecord<>("my-topic", "key1", "value1");

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("发送成功，分区：" + metadata.partition() 
                + "，偏移量：" + metadata.offset());
        }
    }
});

异步发送兼顾了性能和可靠性。你想想看，生产环境每秒可能要发几万条消息，如果用同步方式，吞吐量直接被打回原形。用异步回调，既能知道发送结果，又不阻塞主线程。

三种方式的对比，我整理了一张表：

3.3 分区策略：消息到底去了哪个分区？

消息发到Topic之后，到底落到哪个分区？这取决于分区策略。Kafka默认提供了几种策略，我一个个说。

3.3.1 默认分区器（DefaultPartitioner）

默认分区器的逻辑很简单：

• 如果key不为null：对key做哈希，然后对分区数取模。同一个key永远落到同一个分区。
• 如果key为null：使用轮询（Round-Robin）方式，均匀分布到各个分区。

这里有个坑。我记得有一次排查数据倾斜问题，发现某个分区数据量特别大。后来一看，是因为业务方用了时间戳作为key，而时间戳的哈希值分布不均匀。所以，选key的时候要小心，尽量选分布均匀的字段。

3.3.2 粘性分区策略（Sticky Partitioner）

从Kafka 2.4版本开始，默认分区器改成了粘性分区策略。说白了就是：在缓冲区满了之前，尽量把消息发到同一个分区。这样做的好处是减少网络请求次数，提升吞吐量。

你想想看，如果每条消息都轮询到不同分区，那每次都要单独发一个请求。粘性分区把一批消息攒到一起发，性能自然就上去了。

3.4 自定义分区器：当默认策略不够用时

默认分区器能满足大部分场景，但有时候业务逻辑比较特殊，比如：

• 想把某个用户的订单都发到同一个分区
• 想根据消息中的某个字段做路由
• 想实现加权分配（某些分区多存一些）

这时候就需要自定义分区器了。实现起来很简单，实现 Partitioner 接口就行：

public class OrderIdPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 获取分区数
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        // 如果key为null，随机分配
        if (keyBytes == null) {
            return ThreadLocalRandom.current().nextInt(numPartitions);
        }

        // 根据订单ID的后四位取模
        String orderId = (String) key;
        String suffix = orderId.substring(orderId.length() - 4);
        int partition = Math.abs(suffix.hashCode()) % numPartitions;

        System.out.println("订单 " + orderId + " 分配到分区 " + partition);
        return partition;
    }

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

    @Override
    public void configure(Map<String, ?> configs) {
        // 读取配置
    }
}

然后在生产者配置里指定：

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

就这么简单。我在做电商实时ETL时，就用了类似的分区器，把同一个订单的所有变更事件都路由到同一个分区，这样Flink消费时就能保证顺序性。

3.5 本章知识体系总览

下面这张图把生产者API的核心流程串起来了，你可以对照着回顾一下：