3. Producer API 详解：同步发送与异步发送、回调机制、分区策略、消息序列化与自定义序列化器

Kafka Producer API 是消息生产者的核心编程接口，掌握其发送模式、回调机制、分区策略以及序列化原理，是构建高吞吐、高可靠消息系统的关键。本节将深入剖析 Producer API 的四大核心模块，并通过代码示例与架构图帮助理解其内部协作流程。

3.1 同步发送与异步发送

Kafka Producer 提供两种消息发送模式：同步发送（Sync Send） 和 异步发送（Async Send）。两种模式在调用方式、阻塞行为与可靠性保障上存在显著差异。

3.1.1 同步发送代码示例

// 同步发送：阻塞直到收到 Broker 确认
ProducerRecord<String, String> record =
    new ProducerRecord<>("orders", "order-001", "{\"id\":1,\"amount\":99.9}");
try {
    RecordMetadata metadata = producer.send(record).get();  // 阻塞
    System.out.printf("同步发送成功: topic=%s, partition=%d, offset=%d%n",
        metadata.topic(), metadata.partition(), metadata.offset());
} catch (ExecutionException | InterruptedException e) {
    System.err.println("同步发送失败: " + e.getMessage());
}

3.1.2 异步发送代码示例

// 异步发送：立即返回，通过回调处理结果
ProducerRecord<String, String> record =
    new ProducerRecord<>("orders", "order-002", "{\"id\":2,\"amount\":199.0}");
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.printf("异步发送成功: partition=%d, offset=%d%n",
            metadata.partition(), metadata.offset());
    } else {
        System.err.println("异步发送失败: " + exception.getMessage());
        // 此处可执行重试逻辑或记录死信
    }
});

3.2 回调机制

回调（Callback）是异步发送模式的核心组件，允许 Producer 在消息发送完成后（成功或失败）执行自定义逻辑。Kafka 提供的 Callback 接口仅包含一个方法：onCompletion(RecordMetadata metadata, Exception exception)。

• 成功场景： metadata 非空，exception 为 null，可获取分区、偏移量等信息。
• 失败场景： exception 非空，metadata 为 null，需处理重试或记录错误。
• 线程模型： 回调在 Producer 的 I/O 线程中执行，应避免阻塞操作。
• 顺序保障： 对于同一个分区，回调的执行顺序与消息发送顺序一致。

3.3 分区策略

分区策略决定了消息被发送到目标 Topic 的哪个 Partition。Kafka 提供默认分区器，也支持自定义分区器以满足业务路由需求。

3.3.1 默认分区策略

3.3.2 自定义分区器

实现 org.apache.kafka.clients.producer.Partitioner 接口，重写 partition() 方法。例如，按订单地域将消息路由到特定分区：

public class RegionPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String region = extractRegion((String) key);
        int partitionCount = cluster.partitionCountForTopic(topic);
        // 假设 region 为 "east" 映射到分区 0，"west" 映射到分区 1
        return "east".equals(region) ? 0 : 1 % partitionCount;
    }
    @Override public void close() {}
    @Override public void configure(Map<String, ?> configs) {}
}

3.4 消息序列化与自定义序列化器

Kafka 网络传输要求消息以字节数组形式发送。序列化器负责将 Java 对象转换为字节流，反序列化器负责还原。Producer 端必须配置 key.serializer 和 value.serializer。

3.4.1 内置序列化器

3.4.2 自定义序列化器

当业务对象无法被内置序列化器覆盖时，可实现 org.apache.kafka.common.serialization.Serializer<T> 接口。以下示例将 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 configure(Map<String, ?> configs, boolean isKey) {}
    @Override public void close() {}
}

3.4.3 序列化与分区策略协作流程

3.5 核心参数调优建议