4. Kafka 生产者 API：同步/异步发送、回调机制、分区策略、序列化器

Kafka 生产者（Producer）是数据进入 Kafka 集群的入口。本章深入剖析生产者 API 的核心机制，包括发送模式、回调处理、分区路由与序列化，帮助你在实际项目中构建高吞吐、高可靠的数据生产端。

4.1 生产者发送流程概览

生产者客户端将消息发送到 Broker 之前，会经历序列化、分区选择、缓冲区暂存、批量发送等关键步骤。下图展示了消息从业务对象到网络数据包的完整路径：

4.2 同步发送与异步发送

Kafka 生产者提供两种发送模式，核心区别在于是否阻塞等待 Broker 确认。

// 同步发送：阻塞直到收到 Broker 确认
try {
    RecordMetadata metadata = producer.send(new ProducerRecord<>("orders", key, value)).get();
    System.out.println("发送成功，分区：" + metadata.partition() + "，偏移量：" + metadata.offset());
} catch (ExecutionException | InterruptedException e) {
    System.err.println("发送失败：" + e.getMessage());
}

// 异步发送：立即返回，通过回调处理结果
producer.send(new ProducerRecord<>("orders", key, value), (metadata, exception) -> {
    if (exception == null) {
        System.out.println("发送成功，分区：" + metadata.partition() + "，偏移量：" + metadata.offset());
    } else {
        System.err.println("发送失败：" + exception.getMessage());
        // 此处可加入重试逻辑或死信队列
    }
});

4.3 回调机制

回调（Callback）是异步发送模式下处理发送结果的标准方式。Kafka 提供的 Callback 接口只有一个方法 onCompletion(RecordMetadata metadata, Exception exception)，在消息发送完成后被调用。

• 线程模型：回调在生产者 I/O 线程中执行，应避免在回调中执行耗时操作，否则会阻塞后续消息的发送。
• 异常处理：当 exception 不为 null 时，表示发送失败。常见异常包括 TimeoutException、LeaderNotAvailableException、NotEnoughReplicasException 等。
• 重试策略：回调中不建议直接重试，因为可能引发消息乱序。推荐在回调中记录失败消息到本地缓冲区或死信队列，由独立线程统一重试。

4.4 分区策略

分区策略决定了消息被发送到 Topic 的哪个分区。Kafka 提供默认分区器，也支持自定义分区器。

4.4.1 默认分区策略

4.4.2 自定义分区器

实现 Partitioner 接口，重写 partition() 方法，即可自定义分区逻辑。例如，按业务字段将消息路由到特定分区：

public class OrderPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 假设 value 是 JSON 字符串，提取 "region" 字段
        String jsonValue = (String) value;
        if (jsonValue.contains("\"region\":\"east\"")) {
            return 0; // 东部区域路由到分区 0
        } else if (jsonValue.contains("\"region\":\"west\"")) {
            return 1; // 西部区域路由到分区 1
        }
        // 其他区域使用默认哈希
        return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);
    }

    @Override
    public void close() {}

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

在生产者配置中指定自定义分区器：

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

4.5 序列化器

序列化器负责将 Java 对象转换为字节数组，以便在网络中传输。Kafka 内置了多种序列化器，也支持自定义。

4.5.1 内置序列化器

4.5.2 自定义序列化器

实现 Serializer<T> 接口，重写 serialize() 方法。以下示例将 Order 对象序列化为 JSON 字节：

public class OrderSerializer implements Serializer<Order> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, Order data) {
        if (data == null) return null;
        try {
            return mapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException("序列化 Order 失败", e);
        }
    }

    @Override
    public void close() {}
}

生产者配置中指定 Key 和 Value 的序列化器：

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

4.6 生产者配置核心参数

4.7 本章小结

• 同步发送：调用 send().get()，阻塞等待确认，适合可靠性优先场景。
• 异步发送：调用 send() 并传入回调，不阻塞主线程，适合高吞吐场景。
• 回调机制：在 I/O 线程中执行，应轻量快速，避免阻塞。
• 分区策略：默认按 Key 哈希分区，可自定义分区器实现业务路由。
• 序列化器：内置 String/Integer/Long/ByteArray 等，支持 Avro/Protobuf 及自定义序列化。
• 核心参数：acks、batch.size、linger.ms、compression.type 是调优吞吐与可靠性的关键。