Kafka生产者如何批处理？

Question

假设我有两个经纪人。

我读到Kafka制作人创建的制作人线程等于经纪人的数量。在这种情况下，我将有两个内部线程。

假设我有5个主题，每秒只收到200条消息。Kafka如何进行批处理？

一批大小=30条消息。[topic1=5，topic2=10，topic3=3，topic4=10，topic5=2消息]这些是最重要的消息和相应的主题。

Kafka是如何执行批处理的？

Answer 1

记录累加器负责批量处理发送给Kafka经纪人的值。

- maintains a ConcurrentMap<TopicPartition, Deque<RecordBatch>>
- gets Deque of relevant TopicPartition
- gets last RecordBatch present in Deque, if RecordBatch(bytebuffer bounded by batch.size) is not full appends value to the RecordBatch
- if last RecordBatch is null, no RecordBatch exists for the relevant topic partition hence allocates a new byte buffer
- does a double check locking on last RecordBatch again, incase some other thread might have created the RecordBatch
- if RecordBatch exists, tries appending the value
- if still RecordBatch is null, creates MemoryRecords (backed by byte buffer)
- adds MemoryRecords to RecordBatch
- appends value to RecordBatch ( inside MemoryRecords eventually Byte Buffer )
- adds RecordBatch to Deque

类层次结构：

RecordBatch
 - MemoryRecords
   - ByteBuffer

Answer 2

我读到Kafka制作人创建的制作人线程等于经纪人的数量。在这种情况下，我将有两个内部线程。

不确定，你是从哪里得到这些信息的，但这是不正确的。一个KafkaProducer确实有一个后台线程来异步地向代理写入数据。

配料是如何发生的，很难详细预测。这取决于你的批次。大小（这是一个最大值）。此外，还有linger。ms参数，用于定义在发送数据之前保留数据的时间（即使批次未满）。

更详细地说，将有到托管您写入的分区的所有代理的开放网络连接。此外，批处理是基于分区的——但是，多个批处理可以包含在对代理的单个请求中。