MapR Streams Kafka API 的批处理大小问题

Question

您好，我正在使用Kafka MapRStream接收来自Mapr Streams Topic的事件。

我正在尝试增加我的使用者的批量大小，但我在一个批次中收到的消息不超过30条！

单个事件的大小约为 5000 字节。如果事件较小，我会在一批中获得更多。

这是我对消费者的配置：

public static void main( String[] args ) {
final Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "batchSize");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 50000);
props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 26214400);
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 100 * 1024 * 1024);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);

Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList(TOPIC));
long totalCount = 0;
long start = System.currentTimeMillis();
long countTimesNoMessages = 0;
while (countTimesNoMessages < 10) {
ConsumerRecords<String, String> records = consumer.poll(1000);
totalCount += records.count();
System.out.println(records.count());
if (records.count() == 0) {
countTimesNoMessages++;
}
}
long end = System.currentTimeMillis();
System.out.println((end - start) + " for " + totalCount + " messages");
} 

Answer 1

这些是可能的配置点。

https://mapr.com/docs/61/MapR_Streams/configuration-parameters.html

请注意，fetch.max.bytes是总最大值，所有分区的sum(max.partition.fetch.bytes)不能超过 fetch.max.bytes。

调整max.partition.fetch.bytes是正常的，因此从每个分区轮询超过 64Kb(默认值(，并且还会调整fetch.max.bytes以便max.partition.fetch.bytes正常工作。

您可能不应该将批大小设置得太大。一旦轮询流的请求频率降至每秒几百个左右，您就不太可能获得额外的性能改进，并且在线程失败的情况下更有可能遇到热点问题或大量重做工作。