4. Kafka Java Producer API：核心参数、发送方式与分区策略

好，咱们进入 Producer API 的正题。说实话，Kafka 的 Producer 是我用得最多的客户端，没有之一。很多初学者觉得 Producer 就是 new 一个对象，然后 send 就完事了。但实际生产环境里，参数调不对、分区策略选不好，性能差距能到十倍以上。

今天我就把 Producer 的核心知识点拆开揉碎了讲。你跟着我走一遍，保证能写出靠谱的生产者代码。

4.1 Producer 核心参数：别让默认值坑了你

先看一段最基础的 Producer 初始化代码：

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);

这段代码能跑，但离生产可用还差得远。我列几个你必须要关注的参数：

参数名 默认值 我的建议
acks 1 生产环境用 all，保证不丢数据
retries Integer.MAX_VALUE 配合 retry.backoff.ms 使用，默认100ms
batch.size 16384 (16KB) 根据消息大小调整，我常用32KB或64KB
linger.ms 0 建议设5-20ms，提升吞吐量
buffer.memory 33554432 (32MB) 高吞吐场景建议64MB或128MB
compression.type none 推荐 snappy 或 lz4
max.request.size 1048576 (1MB) 如果消息体大，记得调大

核心原则：Producer 的参数配置，本质是在「吞吐量」「延迟」「可靠性」三者之间做权衡。没有银弹，只有适合你场景的组合。

举个例子，acks=all 意味着 leader 副本要等所有 ISR 副本都确认后才返回。这能保证数据不丢，但延迟会高一些。我做过一个日志采集系统，对延迟不敏感，就用了 acks=all + compression.type=snappy，吞吐量提升了将近40%。

我曾经踩过的坑：有一次线上事故，Producer 一直报 TimeoutException。查了半天，发现是 max.request.size 没改，而业务方发了一条2MB的消息。Kafka 直接拒绝了。从那以后，我养成了一个习惯——上线前先压测一遍消息大小分布。

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

Producer 发送消息，本质上就两种模式。咱们直接看代码：

4.2.1 同步发送（Fire-and-Wait）

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

同步发送就是调了 send() 之后，调用 get() 阻塞等待结果。这种方式简单粗暴，但吞吐量很低。我一般只在测试代码或者对顺序要求极其严格的场景下用。

4.2.2 异步发送（Fire-and-Forget + Callback）

producer.send(new ProducerRecord<>("topic-test", "key", "value"), new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            // 处理发送失败
            exception.printStackTrace();
        } else {
            // 发送成功
            System.out.println("offset: " + metadata.offset());
        }
    }
});

这才是生产环境的主流用法。你想想看，如果每条消息都等确认，那 Producer 就变成串行处理了。异步发送把消息丢进缓冲区，后台线程批量发送，性能能差一个数量级。

我的习惯：异步发送一定要带 Callback。不带 Callback 的 send 就像把消息扔进黑洞——你永远不知道它到底发没发出去。我曾经接手过一个项目，全是不带 Callback 的 send，结果丢了几十万条数据都没人发现。

4.3 回调函数：别只打印日志

Callback 接口只有一个方法 onCompletion，但这里面的门道不少。

先看一个常见的错误写法：

// 错误示范：在回调里做耗时操作
producer.send(record, (metadata, exception) -> {
    // 写数据库、调远程API...
    saveToDatabase(metadata);
});

为什么说这是错的？因为 Callback 是在 Producer 的 I/O 线程里执行的。你在里面做耗时操作，会阻塞后续消息的发送。我见过有人把 Elasticsearch 写入放在 Callback 里，结果 Producer 吞吐量直接掉到原来的十分之一。

正确的做法是：

  1. 回调里只做轻量操作——比如更新计数器、记录日志
  2. 耗时操作丢到线程池——用独立的线程池处理回调结果
  3. 一定要处理异常——exception 不为 null 时，记录日志并考虑重试
// 推荐做法：回调里只做轻量操作
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("发送失败", exception);
        // 可以丢到重试队列
        retryQueue.offer(record);
    } else {
        successCount.incrementAndGet();
    }
});

记住：Callback 是异步的，但它运行在 Producer 的内部线程中。别把它当成业务逻辑的执行入口。

4.4 分区策略：数据该去哪，谁说了算？

Producer 发消息时，怎么决定这条消息去哪个分区？这里面有一套完整的策略。

4.4.1 默认分区器

Kafka 自带的 DefaultPartitioner 逻辑很简单：

  • 如果 key 不为 null：对 key 做哈希，然后对分区数取模。相同 key 的消息永远进同一个分区。
  • 如果 key 为 null：使用粘性分区策略（Sticky Partitioner），把一批消息尽量发到同一个分区，提升批处理效率。

这里有个细节：Kafka 2.4 之后，key 为 null 时不再用轮询，而是用粘性分区。为什么呢？因为轮询会导致每个分区都只有少量消息，batch 很难填满，吞吐量上不去。

4.4.2 自定义分区器

默认策略不够用时，你可以自己写分区器。比如按业务 ID 路由：

public class BusinessPartitioner 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();
        
        // 按业务ID取模，保证同一业务的数据在同一个分区
        String businessId = extractBusinessId(key.toString());
        return Math.abs(businessId.hashCode()) % numPartitions;
    }

    @Override
    public void close() {}

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

然后在 Producer 配置里指定：

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

注意：自定义分区器一定要处理好分区数变化的情况。如果 topic 扩容了，你的取模逻辑可能导致数据倾斜。我一般会在分区器里加一个监控指标，实时观察各分区的数据分布。

4.5 知识体系总览

下面这张图把 Producer API 的核心脉络串起来了，你可以对照着回顾一下：

Kafka Producer API 核心知识体系 KafkaProducer 核心参数配置 发送消息方式 分区策略 关键参数 acks / retries / batch.size linger.ms / buffer.memory compression.type / max.request.size 发送方式 同步：send() + get() 异步：send() + Callback Callback 注意事项 分区策略 DefaultPartitioner Sticky Partitioner 自定义 Partitioner 核心权衡：吞吐量 ↔ 延迟 ↔ 可靠性 没有银弹，只有适合你场景的组合 💡 我的经验： 先确定业务对可靠性的要求，再调吞吐量。不要一上来就追求极致性能。

这张图把 Producer 的三大模块串起来了。你写代码的时候，可以对照着检查：参数配对了没？发送方式选对了没？分区策略合不合理？

← 上一章 📖 返回目录 下一章 →