4. Kafka 生产端实战：Producer API、发送模式、分区策略与消息确认机制

好，咱们今天来聊聊生产端。Kafka 里谁在发消息？Producer。说白了，它就是数据的源头。你想想看，如果源头就出了问题，后面 Consumer 再怎么折腾也是白搭。所以这一章，我带你把这几个核心点彻底吃透：API 怎么用、消息怎么发、发到哪个分区、以及到底怎么才算「发成功」。

4.1 Producer API 基础：从 Hello World 开始

先写一个最简单的 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");

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

这段代码看着简单，但有几个坑。我记得刚带团队时，有个同事把 bootstrap.servers 拼错了，结果连不上集群，查了半天。嗯，这里要注意：bootstrap.servers 不是只连一台，而是用来发现整个集群的。你写一台 broker 的地址，Producer 会自动拉取元数据，找到 topic 的所有分区 leader。

序列化器也很关键。Kafka 只存字节数组，所以 key 和 value 都得序列化。我习惯用 StringSerializer 做快速验证，但生产环境里，Avro 或 Protobuf 才是正经选择。这个后面章节会展开。

4.2 发送模式：同步 vs 异步

Producer 发消息，有两种姿势：同步和异步。说白了，就是「等不等结果」的问题。

4.2.1 同步发送

同步发送，就是调用 send() 后，线程阻塞，直到 broker 返回结果。

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

这里用了 .get()，它会一直等。好处是逻辑简单，能立刻知道结果。坏处呢？吞吐量低。你想想看，每条消息都等，那跟串行有啥区别？

我曾经在一个日志采集项目里用过同步发送，结果 QPS 死活上不去。后来一分析，网络往返时间成了瓶颈。所以同步发送，我只在「必须确认上一条成功才能发下一条」的场景用，比如初始化时创建 topic 的元数据。

4.2.2 异步发送 + 回调

异步才是生产环境的主流。调用 send() 后立即返回，消息进入缓冲区，后台线程批量发送。你只需要注册一个回调，处理成功或失败。

producer.send(new ProducerRecord<>("my-topic", "key", "value"), new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            // 记录日志，重试或告警
            System.err.println("发送失败: " + exception.getMessage());
        } else {
            // 可选：记录发送成功的 offset
        }
    }
});

这里有个细节：回调是在 Producer 的 I/O 线程里执行的。所以回调里千万别做耗时操作，比如写数据库、调远程接口。否则会阻塞发送线程，拖垮吞吐量。我习惯在回调里只做轻量级统计，比如计数器加一，或者把失败消息塞到一个内存队列里，由另一个线程去处理。

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

Kafka 的 topic 可以分成多个分区。消息发到哪个分区，由分区器决定。默认的分区策略是这样的：

• 如果 key 为 null： 使用轮询（Round-Robin）或粘性分区（Sticky Partitioner，2.4+ 版本默认）。粘性分区会尽量把一批消息发到同一个分区，减少网络开销。
• 如果 key 不为 null： 对 key 做哈希，相同 key 的消息永远进同一个分区。这保证了同一个 key 的消息有序。

我举个例子。假设你在做订单系统，用 orderId 做 key。那么同一个订单的所有状态变更消息，都会落到同一个分区。Consumer 消费时，就能保证这个订单的消息是有序的。这个特性在业务里非常实用。

但要注意，如果 key 的基数很小（比如只有 0 和 1 两种值），会导致分区数据严重倾斜。我曾经见过一个案例，key 只有 3 种，结果 10 个分区里 3 个分区数据量巨大，其他 7 个几乎空着。所以设计 key 时，要保证足够的散列粒度。

你也可以自定义分区器。实现 Partitioner 接口，重写 partition() 方法。比如按地域路由，把北京的消息发到分区 0，上海的发到分区 1。

public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String region = extractRegion(key.toString());
        if ("beijing".equals(region)) return 0;
        if ("shanghai".equals(region)) return 1;
        return 2; // 默认
    }
}

自定义分区器虽然灵活，但会增加维护成本。我建议除非有明确的业务需求，否则用默认的就行。

4.4 消息确认机制：acks 参数详解

这是生产端最关键的参数，没有之一。它决定了 Producer 怎么才算「发送成功」。Kafka 提供了三个级别：

我一个个说。

acks=0： 说白了就是「发完拉倒」。Producer 把消息扔到网络缓冲区就算完事。如果网络抖动或 broker 挂了，消息就丢了。但延迟极低。我一般在采集服务器指标时用这个，丢几条 CPU 数据无所谓，但吞吐量要拉满。

acks=1： 这是默认值。Leader 分区收到消息，写入本地日志后，就给 Producer 回 ACK。但如果 Leader 在 Follower 还没同步时就挂了，这条消息就会丢失。大部分业务系统用这个就够了，性能和可靠性比较均衡。

acks=all： 最安全，也最慢。Leader 要等所有 ISR 里的 Follower 都同步完，才返回 ACK。这样即使 Leader 挂了，Follower 上还有数据。但代价是延迟明显增加。我在做支付对账系统时，必须用 acks=all，一条都不能丢。

另外，acks=all 要配合 min.insync.replicas 参数使用。这个参数指定了最少要有几个副本同步成功，才算「所有 ISR 同步完成」。比如 topic 有 3 个副本，min.insync.replicas 设为 2，那么至少 2 个副本同步成功，Producer 才会收到 ACK。如果 ISR 里只剩 1 个副本，Producer 会抛出异常。这个机制防止了「只有一个副本时还傻等」的情况。

4.5 核心流程总览

说了这么多，咱们用一张图把整个生产端的流程串起来。

流程很清晰：应用调用 send()，消息经过序列化、分区路由，进入缓冲区。Sender 线程批量拉取，发到 Broker。Broker 根据 acks 配置决定何时回 ACK。Producer 收到 ACK 后，触发回调。

这里面每个环节都有参数可以调优。比如 batch.size 控制批次大小，linger.ms 控制等待时间，buffer.memory 控制缓冲区总大小。这些参数配合好了，才能榨干 Kafka 的性能。

4.6 小结

这一章我们聊了生产端的四个核心点：API 基础、发送模式、分区策略、acks 确认机制。说白了，就是「怎么发、发到哪、怎么算成功」。这些是 Kafka 生产端最基础也最重要的知识。你想想看，如果连消息都发不对，后面 Consumer 再强也没用。