Kafka源码P2-生产者缓冲区

欢迎来到啾啾的博客🐱。
记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。
有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。

1 引言

继续看Kafka源码，看其是如何批量发送消息的。

2 缓冲区

当调用producer.send(record)时，消息将先到缓冲区，在缓冲区按照目标的Topic-Partition进行组织，满足以条件后随批次发送给Broker。

// KafkaProducer.java
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {// ... 省略了部分代码 ...return doSend(record, callback); // 转交给 doSend 方法
}private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {// ...// 1. 等待元数据更新（如果需要的话）// ...// 2.【核心步骤】调用 RecordAccumulator 的 append 方法RecordAccumulator.RecordAppendResult result = accumulator.append(tp,timestamp, key, value, headers, interceptors, remainingWaitMs);// ...// 3. 唤醒 Sender 线程，告诉他“可能有新活儿干了”this.sender.wakeup();// ...return result.future;
}

我们看一下“缓冲区”RecordAccumulator。

2.1 消息在Partition内有序

RecordAccumulator维护了一个数据结构：

private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches

append代码简化如下：

// RecordAccumulator.java
public RecordAppendResult append(TopicPartition tp, long timestamp, byte[] key, byte[] value, Header[] headers, ...) {// 1. 获取该分区的批次队列 batches中获取，没有则创建Deque<ProducerBatch> dq = getOrCreateDeque(tp);synchronized (dq) { // 对该分区的队列加锁，保证线程安全// 2. 尝试追加到最后一个（当前活跃的）批次中ProducerBatch last = dq.peekLast();if (last != null) {FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, ...);if (future != null) {// 如果追加成功，直接返回return new RecordAppendResult(future, dq.size() > 1 || last.isFull(), false);}}// 3. 如果最后一个批次满了，或者不存在，就需要一个新的批次// 从 BufferPool 申请一块内存，大小由 batch.size 配置决定ByteBuffer buffer = free.allocate(batchSize, maxTimeToBlock);// 4. 创建一个新的 ProducerBatch (货运箱)ProducerBatch batch = new ProducerBatch(tp, memoryRecordsBuilder, now);FutureRecordMetadata future = batch.tryAppend(timestamp, key, value, headers, ...); // 把当前消息放进去// 5. 将新的批次加入到队列的末尾dq.addLast(batch);// ...return new RecordAppendResult(future, ...);}
}

可以看到batches的value类型为Deque，所以生产者可以维护发送时partition内的顺序结构。
但是在网络抖动时这样做还是不够，时序性还是难以保障，所以生产者还有别的配置:
每个连接上允许发送的未确认请求的最大数量

max.in.flight.requests.per.connection

• 当 max.in.flight.requests.per.connection = 1 时，Sender 线程在发送完 Batch-1 后，会阻塞自己，直到 Batch-1 的请求得到响应（成功或失败），它绝不会在此期间发送 Batch-2。这样一来，即使 Batch-1 需要重试，Batch-2 也只能乖乖地在后面排队。这就从根本上杜绝了因重试导致乱序的可能。
• 默认 max.in.flight.requests.per.connection = 5。即它允许 Producer 在还没收到 Batch-1 的 ACK 时，就继续发送 Batch-2、3、4、5。这极大地提升了吞吐量（不用傻等），但牺牲了顺序性。

一般max.in.flight.requests.per.connection还需要与生产者幂等性配合。

enable.idempotence = true

开启幂等后，Producer 会被分配一个唯一的 Producer ID (PID)，并且它发送的每一批消息都会带上一个从0开始递增的序列号。Broker 端会为每个 TopicPartition 维护这个 PID 和序列号。如果收到的消息序列号不是预期的下一个，Broker 就会拒绝它。

// NetworkClient.java// 这个方法判断我们是否可以向某个节点发送更多数据
@Override
public boolean isReady(Node node, long now) {// ... 省略了连接状态的检查 ...// 检查在途请求数是否小于该连接配置的上限return !connectionStates.isBlackedOut(node.idString(), now) &&canSendRequest(node.idString(), now);
}// canSendRequest 方法内部会调用 inFlightRequests.canSendMore()
// InFlightRequests.java
public boolean canSendMore(String nodeId) {// this.requests 是一个 Map<String, Deque<NetworkClient.InFlightRequest>>// 它记录了每个节点上所有在途（已发送但未收到响应）的请求Deque<InFlightRequest> queue = requests.get(nodeId);// 如果队列为空，当然可以发送if (queue == null) {return true;}// 将在途请求数 与 从配置中读到的max.in.flight.requests.per.connection比较// this.maxInFlightRequestsPerConnection 就是你配置的那个值return queue.size() < this.maxInFlightRequestsPerConnection;
}

2.2 批次消息ProducerBatch

可以看到在结构private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches中，批次消息被封装为ProducerBatch。

2.2.1 内存分配

这个类的核心是MemoryRecordsBuilder。
总所周知，频繁地创建和销毁对象，特别是大块的byte[]对GC非常不友好。MemoryRecordsBuilder内部管理者一个巨大的、连续的ByteBuffer。
这个 ByteBuffer 不是每次创建 ProducerBatch 时都 new 出来的。它是在 RecordAccumulator 初始化时(我们在上面的RecordAccumulator中有看到BufferPool)，从一个叫 BufferPool 的内存池中借用 (allocate) 的。当 ProducerBatch 发送完毕，这块内存会归还 (deallocate) 给池子，供下一个 ProducerBatch 复用
当你调用 tryAppend 添加消息时，消息的 key, value 等内容被直接序列化成字节，并写入到这个 ByteBuffer 的末尾。它不是在发送时才做序列化，而是在追加时就完成了。

池化：对于那些需要频繁创建和销毁的、生命周期短暂的、昂贵的对象（如数据库连接、线程、大块内存），一定要使用池化技术。这能极大地降低GC压力，提升系统稳定性。

Redis和Kafka都有共同的高频内存使用的特性，也都设计了预分配和复用。Kafka生产者与其用多少申请多少，不如一次性申请一块大内存，然后通过内部的指针移动（position, limit）来管理这块内存的使用。

2.2.2 线程安全

ProducerBatch 会被多个线程访问：

• 你的业务线程（Producer主线程）：调用 tryAppend() 往里面写数据。
• Sender 线程：检查它是否已满 (isFull)、是否超时 (isExpired)，并最终把它发送出去。

ProducerBatch 内部有一个精密的“状态机”，并用 volatile 和 synchronized 保护。

// ProducerBatch.java (简化后)
private final List<Thunk> thunks;
private final MemoryRecordsBuilder recordsBuilder;// 【关键状态】volatile 保证了多线程间的可见性
private volatile boolean closed; 
private int appends; // 记录追加次数public FutureRecordMetadata tryAppend(...) {// 【关键检查】在方法入口处检查状态，快速失败if (this.closed) {return null; }// ... 将消息写入 recordsBuilder ...// ...
}// 这个方法会被 Sender 线程调用
public void close() {this.closed = true;
}// 当批次被确认后，由 Sender 线程调用
public void done(long baseOffset, long logAppendTime, RuntimeException exception) {// for-each 循环是线程安全的，因为 thunks 列表在 close 之后就不再被修改for (Thunk thunk : this.thunks) {try {// 【核心】执行每个 send() 调用对应的回调函数thunk.callback.onCompletion(metadata, exception);} catch (Exception e) {// ...}}
}

总的来说是职责分离+最小化锁的设计以保证线程安全。

3 发送消息Sender

Sender 是一个实现了 Runnable 接口的类，它在一个独立的线程里无限循环，最终发送消息。

// Sender.java
public void run() {while (running) {try {runOnce();} catch (Exception e) {// ...}}
}void runOnce() {// ...// 1. 【核心】找出所有可以发送的批次// linger.ms 决定了可以等待的最长时间RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);// 2. 如果有准备好的节点（分区），就发送它们if (!result.readyNodes.isEmpty()) {// ...// 从累加器中“榨干”所有准备好的批次Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, ...);// ...// 将批次转换成网络请求并发送sendProducerData(now, ... , batches);}// ...
}

RecordAccumulator.ready() 方法是决定何时发送的关键。它会遍历所有的 ProducerBatch，满足以下任意一个条件的批次，就会被认为是 “ready”（准备就绪）：

1. 批次已满：批次大小达到了 batch.size。
2. 等待超时：批次从创建到现在，等待的时间超过了 linger.ms。
3. 其他原因：比如 Producer 被关闭，或者有新的 Producer 加入导致需要立即发送等。

Sender 的工作模式是：
不断地问accumulator.ready()有到linger.ms时间的或者装满batch.size的批次没有。然后依据节点列表，通过NetworkClient发送ProducerBatch到Kafka Broker。

4 总结

Kafka Producer 在客户端内部通过 RecordAccumulator 维护了一个按 TopicPartition 分类的内存缓冲区。当用户调用 send() 方法时，消息并不会立即发送，而是被追加到对应分区的某个 ProducerBatch 中。一个独立的 Sender 线程在后台运行，它会持续检查 RecordAccumulator 中的批次，一旦某个批次满足了“大小达到 batch.size”或“等待时间超过 linger.ms”这两个条件之一，Sender 线程就会将这个批次以及其他所有准备好的批次一同取出，打包成一个请求，通过网络一次性发送给 Broker，从而实现批量发送，极大地提升了吞吐能力。

这个设计是经典的 “空间换时间” 和 “攒一批再处理” 的思想，通过牺牲一点点延迟（linger.ms），换取了巨大的吞吐量提升。理解了这个机制，你就能更好地去配置 batch.size 和 linger.ms 这两个核心参数，以平衡你的业务对吞吐和延迟的需求。