How to consume from Kafka topic in multiple goroutines, using Sarama?

How to consume from Kafka topic in multiple goroutines, using Sarama?

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.

Solutions/Answers:

Solution 1:

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)

}

References

Loading...