📚 课程目录第4章生产者API详解

4、生产者API详解：同步发送与异步发送、回调函数、分区策略、消息序列化与反序列化

好，咱们进入生产者 API 的核心地带。这部分内容，说白了就是 Kafka 数据流水线的入口控制台。你写的所有数据，都得经过这里才能进到 Broker。我见过不少团队，代码能跑，但一上生产就丢消息、乱分区、性能拉胯——问题基本都出在这几个环节上。

核心要点：生产者 API 的四大支柱——发送模式、回调、分区、序列化。掌握它们，你就能精准控制每一条消息的命运。

4.1 同步发送 vs 异步发送：你选哪个？

Kafka 生产者发送消息，本质上就两种方式：同步和异步。很多人觉得「异步肯定比同步好」，嗯，这话只说对了一半。

4.1.1 同步发送

同步发送，就是调用 send() 之后，线程卡住，等 Broker 确认了再继续。代码长这样：

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

注意那个 .get()，它把 Future 转成了阻塞调用。我个人习惯在两种场景下用同步发送：

关键业务数据：比如订单创建、支付回调，丢一条可能就出事故。
系统启动初始化：需要确保某些配置消息已经落盘，后续流程才能继续。

我曾经在一个日志采集系统里全用同步发送，结果高峰期生产者线程全部阻塞，CPU 打满，消息堆积到内存溢出。同步发送的代价就是吞吐量直线下降——每次都要等网络往返。

4.1.2 异步发送

异步发送才是生产环境的主力。你只管把消息丢出去，Kafka 的发送线程会在后台批量处理。代码更简洁：

ProducerRecord<String, String> record = 
    new ProducerRecord<>("my-topic", "key", "value");
producer.send(record);  // 不阻塞，直接返回

为什么推荐异步？你想想看，Kafka 生产者的核心优势之一就是批量发送。异步模式下，多条消息可以在内存里攒成一个 Batch，一次网络请求发出去，吞吐量能差出几个数量级。

对比维度	同步发送	异步发送
吞吐量	低（每次等待确认）	高（批量发送）
延迟	高（阻塞等待）	低（立即返回）
可靠性	高（能立即感知失败）	依赖回调机制
适用场景	关键业务、初始化	日志、监控、常规数据流

4.2 回调函数：异步世界的「保险丝」

异步发送最大的隐患是什么？消息丢了你不知道。所以回调函数就是那个「保险丝」——发送完成后，无论成功还是失败，它都会通知你。

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            // 发送失败，记录日志或重试
            System.err.println("发送失败: " + exception.getMessage());
            // 这里可以写重试逻辑，或者把消息写入死信队列
        } else {
            // 发送成功，metadata 里包含分区和偏移量
            System.out.println("成功发送到分区 " + metadata.partition() 
                + "，偏移量 " + metadata.offset());
        }
    }
});

我在项目中遇到过一个问题：回调里直接打印日志，结果日志量太大把磁盘打满了。后来我改成在回调里只做轻量操作——比如更新一个计数器，或者把失败消息塞进一个内存队列，由另一个线程专门处理重试。

避坑指南：回调函数是在生产者 I/O 线程里执行的。千万别在回调里做耗时操作（比如写数据库、发 HTTP 请求），否则会阻塞发送线程，拖垮整个生产者的吞吐量。

4.3 分区策略：数据该去哪就去哪

Kafka 的 Topic 分成多个 Partition，消息到底落到哪个分区？这就是分区策略干的事。默认策略其实挺聪明，但有时候你得自己说了算。

4.3.1 默认分区器

默认情况下，Kafka 的分区逻辑是这样的：

有 key：对 key 做哈希，然后对分区数取模。同一个 key 永远进同一个分区。
无 key：使用轮询（Round-Robin）或者粘性分区（Sticky Partitioner，2.4 版本后默认），尽量均匀分布。

粘性分区是个好东西。它会把一批消息先攒到同一个分区，等 Batch 满了再换下一个。这样能提高 Batch 的填充率，减少网络请求次数。我记得刚升级到 2.4 时，同样的流量，生产者 CPU 使用率降了 15%。

4.3.2 自定义分区策略

有时候默认策略满足不了业务。比如你希望所有「北京」用户的数据进分区 0，「上海」的进分区 1。这时候就得自己写分区器：

public class CityPartitioner implements Partitioner {
    private Map<String, Integer> cityToPartition = new HashMap<>();
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 假设 key 是城市名
        String city = (String) key;
        return cityToPartition.getOrDefault(city, 0);
    }
    
    @Override
    public void close() {}
    
    @Override
    public void configure(Map<String, ?> configs) {
        // 从配置里读取城市-分区映射
        cityToPartition.put("北京", 0);
        cityToPartition.put("上海", 1);
        cityToPartition.put("广州", 2);
    }
}

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

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

我曾经见过一个团队自定义分区器里写了数据库查询，结果每次发送消息都查一次库，生产者直接变成性能瓶颈。自定义分区器一定要轻量、无状态、可快速计算。

4.4 消息序列化与反序列化：数据格式的「翻译官」

Kafka 网络传输的是字节数组。你的 Java 对象怎么变成字节？序列化。从字节怎么变回对象？反序列化。这步看似简单，但坑最多。

4.4.1 内置序列化器

Kafka 自带了一些常用的：

类型	序列化器	反序列化器
String	StringSerializer	StringDeserializer
Integer	IntegerSerializer	IntegerDeserializer
Long	LongSerializer	LongDeserializer
ByteArray	BytesSerializer	BytesDeserializer

配置方式很简单：

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

4.4.2 自定义序列化：慎用！

如果你要发送的不是 String 或数字，而是自定义对象，可以自己写序列化器。但我个人强烈建议：别自己写。为什么？

版本兼容性难搞：对象字段变了，老数据就反序列化不了。
跨语言困难：Java 序列化的字节，Python 消费者读不懂。
性能不一定好：Java 原生序列化又慢又占空间。

真要传复杂对象，用 Avro、Protobuf、JSON 这些通用格式。Kafka 官方提供了 Avro 序列化器，配合 Schema Registry 使用，既能保证兼容性，又能跨语言。

// 使用 Avro 序列化器（需要引入依赖）
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

我的经验：生产环境里，80% 的消息用 String 或 JSON 就够了。别为了「面向对象」而引入复杂的序列化方案。简单的东西不容易出错。

4.5 知识体系总览

下面这张图把生产者 API 的核心脉络串起来了。你可以看到，一条消息从业务代码到 Broker，经历了哪些关卡：

从这张图可以看得很清楚：业务代码先经过序列化，把对象变成字节；然后分区器决定去哪个分区；最后选择同步或异步方式发送到 Broker。每一步都有讲究，每一步都可能出问题。

嗯，生产者 API 的核心内容就这些。记住一个原则：异步是常态，回调是保障，分区要合理，序列化要通用。把这四点吃透了，你的数据流水线入口就稳了。