我如何知道要使用 Kafka 中的哪个分区?
Golang不知道大家是否熟悉?今天我将给大家介绍《我如何知道要使用 Kafka 中的哪个分区?》,这篇文章主要会讲到等等知识点,如果你在看完本篇文章后,有更好的建议或者发现哪里有问题,希望大家都能积极评论指出,谢谢!希望我们能一起加油进步!
我在 kafka 代理中有一个包含 3 个分区的主题 1 个分区有消息,2 个分区为空,我如何知道一次调用要消耗哪个分区?
首先,我分配一个分区等于 kafka.partitionany
的 topicpartition,但该值始终返回 -1
所以我必须手动使用计数器,当我成功从一个分区消费但消息为空时,然后 count++
并从下一个开始,直到找到消息
for{
partitions = append(partitions, kafka.TopicPartition{
Topic: &topic,
Partition: partition,
Offset: offSet,
Error: err,
})
err = c.Assign(partitions)
if err != nil {
return err
}
// retrieve message
ev, err := c.Poll(-1)
if err != nil {
return err
}
// if no message, check the next partition
if ev == nil{
partition++
}else{
break
}
}
前两轮不返回任何消息,但要等待第三轮返回,有什么办法可以自动检测哪个分区存储着未消费的消息?
如果没有其他办法,kafka可以为我做循环路由吗?或者我必须自己记录计数器
谢谢! :)
解决方案
您绝对应该使用 subscribe() 方法,然后调用 poll()。 如果有可用记录,您将获得包含一条或多条记录(与不同分区关联)的记录答案。
处理记录后,您可以使用记录中包含的元数据(主题、分区、偏移量等)手动提交(如果您使用 enable.auto.commit = false
)。
雅尼克
今天关于《我如何知道要使用 Kafka 中的哪个分区?》的内容介绍就到此结束,如果有什么疑问或者建议,可以在编程网公众号下多多回复交流;文中若有不正之处,也希望回复留言以告知!
免责声明:
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341