go - 我如何知道在 Kafka 中使用哪个分区？

Question

我在 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能不能帮我做round-robin routing？或者我必须自己记录计数器

谢谢! :)

Answer 1

您绝对应该使用 subscribe() 方法，然后调用 poll()。 如果有可用记录，您将获得包含一个或多个记录(与不同分区关联)的记录答案。

处理记录后，您可以使用记录中包含的元数据(主题、分区、偏移量等)手动提交(如果您使用 enable.auto.commit = false) .

亚尼克