发送大量消息 Kafka Producer

Question

我正在使用卡夫卡。

我有一个包含 10k jsons 的列表。

目前我发送 Jsons 如下：

for(int i=0 ;i< jsonList.size(); i++){
     ProducerRecord<K,V> record = new ProducerRecord(topic, jsonList[i]);
     producer.send(record);
}

发送每条消息。

我想将列表发送到 kafka 并使 kafka 在 json 之后将其发送 json（不是包含所有 json 字符串的一条消息），例如：

ProducerRecord<K,V> record = new ProducerRecord(topic, jsonList);
producer.send(record);

我该怎么做？

谢谢

Answer 1

正式通过使用KafkaProducer，producerRecord你不能这样做，但你可以通过配置一些属性来做到这一点ProducerConfig

来自文档生产者的batch.size将记录批处理为发送到同一分区的请求并立即发送

每当多条记录被发送到同一个分区时，生产者将尝试将记录一起批处理到更少的请求中。这有助于提高客户端和服务器的性能。此配置以字节为单位控制默认批处理大小。不会尝试批量处理大于此大小的记录。

linger.ms此设置用于生产者的延迟时间，为生产者保留一些时间，以便在此期间的所有请求都将被批量发送，但batch.size是此上限，如果生产者获得足够的批量大小，它将忽略此属性并将批处理消息发送到 kafka

生产者将在请求传输之间到达的所有记录组合成单个批处理请求。此设置通过添加少量人为延迟来实现此目的 - 也就是说，生产者不会立即发送记录，而是等待给定的延迟以允许发送其他记录，以便可以将发送一起批处理。此设置给出了批处理延迟的上限：一旦我们获得了一个分区的batch.size记录，无论此设置如何，它都会立即发送。