我使用https://github.com/Shopify/sarama与Kafka进行交互。我有一个主题,例如100个分区。我有部署在1台主机上的应用程序。因此,我想在多个goroutine中使用该主题。

我看到了这个示例-https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go,我们可以在其中看到如何在特定使用者组中创建使用者。

因此,我的问题是,我应该创建多个此类消费者,还是Sarama中有一些设置,可以在其中设置所需数量的消费者goroutine。

P.S. I看到了这个问题-https://github.com/Shopify/sarama/issues/140-但没有答案,如何创建MultiConsumer

分析解答

此示例显示了一个完全正常工作的控制台应用程序,该应用程序可以使用一个主题中的所有分区,每个主题创建一个goroutine:

https://github.com/Shopify/sarama/blob/master/tools/kafka-console-consumer/kafka-console-consumer.go

它链接在您在问题中发布的主题的末尾。

它基本上可以创建一个消费者:

c, err := sarama.NewConsumer(strings.Split(*brokerList, ","), config)

然后获取所需主题的所有分区:

func getPartitions(c sarama.Consumer) ([]int32, error) {
    if *partitions == "all" {
        return c.Partitions(*topic)
    }
...

然后,为每个分区创建一个PartitionConsumer,并在不同的goroutine中从每个分区进行消费:

for _, partition := range partitionList {
    pc, err := c.ConsumePartition(*topic, partition, initialOffset)
    ....

    wg.Add(1)
    go func(pc sarama.PartitionConsumer) {
        defer wg.Done()
        for message := range pc.Messages() {
            messages <- message
        }
    }(pc)

}