使用Scala的Kafka Producer无法进行批处理

Question

我正在用Scala编写一个制作人，我想做批处理。批处理的工作方式是，它应该将消息保持在队列中，直到消息已满，然后将所有消息一起发布到主题上。但不知怎么的，它不起作用了。从我开始发送消息的那一刻起，它就开始一个接一个地发布消息。有人知道如何在Kafka Producer中使用批处理吗。

val kafkaStringSerializer = "org.apache.kafka.common.serialization.StringSerializer"
      val batchSize: java.lang.Integer = 163840
      val props = new Properties()
      props.put("key.serializer", kafkaStringSerializer)
      props.put("value.serializer", kafkaStringSerializer)
      props.put("batch.size", batchSize);
      props.put("bootstrap.servers", "localhost:9092")

      val producer = new KafkaProducer[String,String](props)

      val TOPIC="topic"
      val inlineMessage = "adsdasdddddssssssssssss"

      for(i<- 1 to 10){
        val record: ProducerRecord[String, String] = new ProducerRecord(TOPIC, inlineMessage )
        val futureResponse: Future[RecordMetadata] =  producer.send(record)
        futureResponse.isDone
        println("Future Response ==========>" + futureResponse.get().serializedValueSize())
      }

Answer 1

您正在同步向Kafka服务器生成数据。意思是，当你呼叫制作人时。发送带有未来响应的。get，只有在数据存储在Kafka服务器中后才会返回。

将响应存储在单独的列表中，并在循环的之外调用FutureResponse.get。

在默认配置下，Kafka支持批处理，请参见linger。ms和批处理。尺寸

List<Future<RecordMetadata>> responses = new ArrayList<>();
for (int i=1; i<=10; i++) {
    ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, inlineMessage);
    Future<RecordMetadata> response = producer.send(record);
    responses.add(response);
}

for (Future<RecordMetadata> response : responses) {
    response.get(); // verify whether the message is sent to the broker.
}

Answer 2

你必须在道具中设置linger.ms

默认情况下，它为零，这意味着如果可能的话，消息会立即发送。您可以增加它（例如100），以便批量发生——这意味着更高的延迟，但更高的吞吐量。

批处理。大小是最大值：如果在逗留之前达到该值。ms已通过，数据将在不等待更多时间的情况下发送。

要查看实际发送的批，您需要配置日志记录（批处理在后台线程上完成，您将无法查看使用producer api完成的批处理-您无法发送或接收批处理，只能发送记录并接收其响应，通过批处理与代理的通信在内部完成）

首先，如果尚未完成，请绑定一个log4j属性文件（Dlog4j.configuration=file:path/to/log4j.properties）

log4j.rootLogger=WARN, stderr
log4j.logger.org.apache.kafka.clients.producer.internals.Sender=TRACE, stderr

log4j.appender.stderr=org.apache.log4j.ConsoleAppender
log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
log4j.appender.stderr.Target=System.err

例如，我会收到

TRACE Sent produce request to 2: (type=ProduceRequest, magic=1, acks=1, timeout=30000, partitionRecords=({test-1=[(record=LegacyRecordBatch(offset=0, Record(magic=1, attributes=0, compression=NONE, crc=2237306008, CreateTime=1502444105996, key=0 bytes, value=2 bytes))), (record=LegacyRecordBatch(offset=1, Record(magic=1, attributes=0, compression=NONE, crc=3259548815, CreateTime=1502444106029, key=0 bytes, value=2 bytes)))]}), transactionalId='' (org.apache.kafka.clients.producer.internals.Sender)

这是一批2个数据。批处理将包含发送到同一代理的记录

然后，玩batch.size和linger.ms看看区别。请注意，记录包含一些开销，因此1000的batch.size不会包含10条大小为100的消息

注意，我没有找到说明所有记录器及其功能的文档（比如log4j.logger.org.apache.kafka.clients.producer.internals.Sender）。您可以在rootLogger上启用DEBUG/TRACE并找到所需的数据，或者浏览代码