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 中的哪个分区?》的内容介绍就到此结束,如果有什么疑问或者建议,可以在编程网公众号下多多回复交流;文中若有不正之处,也希望回复留言以告知!