3. Producer 发送原理：分区器、拦截器、序列化器、缓冲区与批次发送机制

Kafka Producer 的发送链路并非将单条消息直接发往 Broker，而是一条由 拦截器 → 序列化器 → 分区器 → 缓冲区 → Sender 线程 构成的异步管道。理解这条管道的每个环节，是诊断堆积、优化吞吐与延迟的前提。

3.1 发送链路全景

Producer 在 send() 调用后，消息会依次经过以下五个核心组件，最终被批量发送到目标 Broker：

1. 拦截器链（Interceptor Chain）：在发送前和发送后回调中修改或记录消息。
2. 序列化器（Serializer）：将 Key 和 Value 从 Java 对象转为字节数组。
3. 分区器（Partitioner）：根据 Key 或自定义策略计算目标分区编号。
4. 缓冲区（RecordAccumulator）：将消息暂存到对应分区的双端队列中，等待打包。
5. Sender 线程：从缓冲区拉取已准备好的批次，通过网络层发送到 Broker。

3.2 分区器（Partitioner）

分区器决定消息写入 Topic 的哪个分区。默认策略基于 Key 的哈希值取模，但生产环境中常需自定义分区逻辑以解决数据倾斜。

3.3 拦截器（Interceptor）

拦截器允许在消息发送前和发送后回调中插入自定义逻辑。典型用途包括：消息审计、格式转换、动态修改 Topic、采集发送指标。

// 自定义拦截器示例
public class MetricsInterceptor implements ProducerInterceptor<String, String> {
    private final AtomicLong successCount = new AtomicLong(0);
    private final AtomicLong failureCount = new AtomicLong(0);

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 发送前：可修改 record 的 value 或 headers
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            successCount.incrementAndGet();
        } else {
            failureCount.incrementAndGet();
        }
    }

    @Override
    public void close() {
        // 上报指标
        System.out.println("Success: " + successCount.get() + ", Failure: " + failureCount.get());
    }

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

3.4 序列化器（Serializer）

序列化器将 Key 和 Value 从 Java 对象转为字节数组。Kafka 内置了 StringSerializer、ByteArraySerializer、IntegerSerializer 等。自定义序列化器需实现 Serializer<T> 接口。

3.5 缓冲区（RecordAccumulator）与批次发送

缓冲区是 Producer 实现异步批量发送的核心。每个分区对应一个双端队列（Deque），队列中存放 批次（Batch）。Sender 线程不断从队列头部拉取已准备好的批次进行发送。

3.5.1 缓冲区核心参数

3.5.2 批次发送流程

1. 追加消息：消息序列化后，根据分区找到对应的 Deque，尝试追加到队尾的 Batch 中。
2. 创建新批次：若当前 Batch 已满（达到 batch.size）或不存在，则从 BufferPool 申请新内存创建 Batch。
3. 等待与唤醒：若缓冲区总内存不足，send() 阻塞直到有空间释放或超时。
4. Sender 轮询：Sender 线程循环检查每个分区的 Deque 头部 Batch，若 Batch 已满或超过 linger.ms，则将其移出并封装为 ProduceRequest。
5. 发送与回调：请求通过 NetworkClient 发送到 Leader Broker，收到响应后触发回调。

3.6 关键参数调优建议