douzhigan1687
2019-07-04 17:49
浏览 1.4k
已采纳

如何使用Sarama在多个goroutine中从Kafka主题消费?

I use https://github.com/Shopify/sarama for interaction with Kafka. I have a topic with, for example, 100 partitions. I have application, which is deployed on 1 host. So, I want to consume from this topic in multiple goroutines.

I see this example - https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go , in which we can see, how to create consumer in specific consumer group.

So, my question is, should I create multiply such consumers, or there is some setting in Sarama, where I can set up needed number of consumer goroutines.

P.S. I see this question - https://github.com/Shopify/sarama/issues/140 - but there is no answer, how to create MultiConsumer.

图片转代码服务由CSDN问答提供 功能建议

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

我看到了此示例- https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go ,在其中我们可以看到如何在特定的消费者群体中创建消费者。

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

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

  • 写回答
  • 好问题 提建议
  • 关注问题
  • 收藏
  • 邀请回答

1条回答 默认 最新

  • doutuanxiao4619 2019-07-04 19:02
    已采纳

    This example shows a fully working console application which can consume for all partitions in a topic creating one goroutine per partition:

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

    It is linked at the end of the thread you posted in your question.

    It basically creates one consumer:

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

    Then gets all the partitions for the desired topic:

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

    Then for each partition it creates a PartitionConsumer and consumes from each partition in a different 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)
    
    }
    
    已采纳该答案
    评论
    解决 无用
    打赏 举报

相关推荐 更多相似问题