如何解决消耗多个分区的kafka批处理
我知道Kafka可以批量提取事件。 我正在尝试了解这种情况:
- 我为一个主题划分了4个分区
- 我有1个使用者,Kafka为其分配了全部4个分区。
- 让我们假设从Kafka提取的每批Kafka客户端都是5条消息。
我想在这里理解的是,如果1个批次中的事件全部来自同一分区,然后循环到下一个分区批次。还是批处理本身已经包含来自不同分区的事件?
解决方法
我无法给您确切的答案,但发现它很有趣,可以对其进行测试。
为此,我创建了一个具有四个分区的主题,并使用kafka-producer-perf-test
命令行工具向该主题生成了一些消息。由于性能测试工具根本不会创建任何键,因此消息将以循环方式写入主题分区。
kafka-producer-perf-test --topic test --num-records 1337 --throughput -1 --record-size 128 --producer-props key.serializer=org.apache.kafka.common.serialization.StringSerializer --producer-props value.serializer=org.apache.kafka.common.serialization.StringSerializer --producer-props bootstrap.servers=localhost:9092
然后,我使用配置max_poll_records=5
创建了一个简单的KafkaConsumer来匹配您的问题。使用者只需打印出所消耗的每条消息的偏移量和分区:
Integer counter = 0;
// consume messages with `poll` call and print out results
try(KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(settings)) {
consumer.subscribe(Arrays.asList(topic));
while (true) {
System.out.printf("Batch = %d\n",counter);
ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String,String> record : records) {
System.out.printf("offset = %d,partition = %d\n",record.offset(),record.partition());
}
counter += 1;
}
}
回答您的问题的结果是,使用者在移至另一个分区之前尝试从一个分区中获取尽可能多的数据。仅在使用了来自分区1
的所有消息但未达到max_poll_records的限制5的情况下,它又添加了来自分区2
的两个消息。
这里有一些印刷品可以使您更好地理解。
Batch = 0
offset = 310,partition = 0
offset = 311,partition = 0
offset = 312,partition = 0
offset = 313,partition = 0
offset = 314,partition = 0
Batch = 1
offset = 315,partition = 0
offset = 316,partition = 0
offset = 317,partition = 0
offset = 318,partition = 0
offset = 319,partition = 0
# only offsets with partition 0
Batch = 45
offset = 525,partition = 0
offset = 526,partition = 0
offset = 527,partition = 0
offset = 528,partition = 0
offset = 529,partition = 0
Batch = 46
offset = 728,partition = 1
offset = 729,partition = 1
offset = 730,partition = 1
offset = 731,partition = 1
offset = 732,partition = 1
# only offsets with partition 1
Batch = 86
offset = 928,partition = 1
offset = 929,partition = 1
offset = 930,partition = 1
offset = 931,partition = 1
offset = 932,partition = 1
Batch = 87
offset = 465,partition = 2
offset = 466,partition = 2
offset = 933,partition = 1
offset = 934,partition = 1
offset = 935,partition = 1
Batch = 88
offset = 467,partition = 2
offset = 468,partition = 2
offset = 469,partition = 2
offset = 470,partition = 2
offset = 471,partition = 2
## and so on
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。