生产者基础：Producer API使用、消息发送方式（同步/异步）、acks与retries配置

好，咱们进入生产者的世界。Kafka 生产者，说白了就是那个往管道里塞数据的角色。我见过不少新手，上来就写个 Producer.send()，然后数据丢了都不知道怎么回事。今天咱们就把这块掰开揉碎，讲清楚。

Producer API 核心组件

先看一段最基础的生产者代码。我习惯从最简单的开始，再慢慢加配置。

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

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

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

这里有几个关键点：

  • bootstrap.servers：别只配一个 broker。我踩过坑，单点挂了整个集群连不上。至少配两三个。
  • 序列化器：Kafka 只传字节。你传对象？得自己转。StringSerializer 是最常用的，但生产环境我更喜欢用 Avro 或 Protobuf。
  • ProducerRecord：消息本体。包含 topic、partition（可选）、key、value。key 为空时，消息会轮询发到各分区。
我的习惯：key 尽量别空。有 key 才能保证相同 key 的消息进同一个分区，顺序才有保障。日志类数据可以不要 key，但业务数据我建议都带上。

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

很多人以为 producer.send() 是同步的。其实不是。它默认是异步的，返回一个 Future 对象。咱们来看看区别。

异步发送（默认）

producer.send(new ProducerRecord<>("topic", "key", "value"),
    (metadata, exception) -> {
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
        } else {
            System.out.println("发送成功, offset: " + metadata.offset());
        }
    });

异步发送不阻塞主线程。回调函数在发送完成后执行。我一般在日志采集场景用这个，吞吐量高。

同步发送

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

调用了 .get()，就变成同步了。线程会卡在这里，直到 broker 返回结果。你想想看，如果网络延迟高，这得卡多久？

我曾经：在一个支付对账系统里用了同步发送，结果高峰期 producer 线程池被打满，整个应用响应变慢。后来改成异步 + 回调重试，问题才解决。同步发送适合对可靠性要求极高、流量不大的场景。

acks 配置：到底等几个确认？

acks 是生产者的核心配置。它决定了 producer 要等多少个 broker 确认，才算发送成功。直接看对比：

acks 值 含义 可靠性 延迟 适用场景
0 不等待确认，发完就算成功 最低 最低 日志、监控指标，丢几条无所谓
1 等待 leader 确认写入 中等 中等 大部分业务场景，默认值
all（或 -1） 等待 leader + 所有 ISR 副本确认 最高 最高 金融、订单、支付等关键数据

我个人习惯：

  • 日志类数据用 acks=0，性能优先。
  • 普通业务用 acks=1，够用了。
  • 核心交易用 acks=all，配合 min.insync.replicas=2，保证至少两个副本有数据。

注意：acks=all 不代表数据 100% 不丢。如果所有 ISR 副本都挂了，数据还是会丢。所以还要配合 replication.factor 和 min.insync.replicas 一起用。

retries 配置：失败后怎么办？

网络抖动、leader 选举，这些都会导致发送失败。retries 就是重试次数。默认是 Integer.MAX_VALUE，但 2.0 之后改成了 0。嗯，这里要注意。

props.put("retries", 3);
props.put("retry.backoff.ms", 100);

重试不是越多越好。我见过一个系统 retries 设了 10 次，backoff 设了 1 秒，结果 broker 挂了 5 分钟，producer 一直在重试，消息堆积了几百万条。最后 OOM 了。

我的建议：retries 设 3-5 次，backoff.ms 设 100-300 毫秒。如果重试还失败，把消息写到本地文件或另一个 topic，后续再处理。别死磕。

核心流程：一张图说清楚

下面这张图，把 producer 发送消息的完整流程画出来了。从序列化到分区，再到缓冲区，最后发到 broker。我每次给团队培训都拿这张图讲。

Kafka Producer 消息发送流程 ProducerRecord topic / key / value 序列化器 Serializer 分区器 Partitioner 缓冲区 RecordAccumulator Sender 线程 后台发送 Broker Leader / Follower 回调 Callback 成功 / 失败处理 关键配置 acks: 0 / 1 / all retries: 重试次数 batch.size: 批次大小 发送方式 异步：send() + Callback 同步：send().get()

流程其实不复杂：

  1. 先创建 ProducerRecord，指定 topic 和消息内容。
  2. 序列化器把 key 和 value 转成字节。
  3. 分区器决定消息去哪个分区。有 key 就 hash，没 key 就轮询。
  4. 消息进缓冲区（RecordAccumulator），攒一批再发。
  5. Sender 线程从缓冲区取数据，批量发送到 broker。
  6. broker 根据 acks 配置返回确认。
  7. 回调函数处理结果。
注意：缓冲区大小默认 32MB。如果发送速度跟不上生产速度，缓冲区满了之后，producer 会阻塞或者抛异常。我建议根据流量调整 buffer.memory 和 max.block.ms。

实战配置清单

最后，给出一份我常用的生产配置。你可以直接拿去用，再根据实际情况微调。

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

// 可靠性配置
props.put("acks", "all");
props.put("retries", 3);
props.put("retry.backoff.ms", 200);
props.put("enable.idempotence", true);  // 幂等性，防止重复

// 性能配置
props.put("batch.size", 16384);        // 16KB
props.put("linger.ms", 5);             // 最多等5ms
props.put("buffer.memory", 67108864);  // 64MB
props.put("compression.type", "snappy"); // 压缩

// 超时配置
props.put("request.timeout.ms", 30000);
props.put("delivery.timeout.ms", 120000);
我的经验：enable.idempotence 设为 true 后，producer 会自动处理重复消息。但注意，它要求 acks=all 且 retries>0。这三个配置是绑定的，少一个都不行。

好了，生产者基础就讲到这里。配置项很多，但核心就三个：acks 管可靠性，retries 管容错，batch.size 和 linger.ms 管性能。你想想看，是不是这个道理？

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