在使用Sarama库进行Kafka消费时，可以通过配置相同的GroupID来实现将多个消费者添加到同一个消费组中。以下是示例代码：
import (
"fmt"
"log"
"github.com/Shopify/sarama"
)
func main() {
brokers := []string{"your-bootstrap-server:9092"}
groupID := "your-group-id"
topic := "your-topic"
config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
consumer, err := sarama.NewConsumer(brokers, config)
if err != nil {
log.Fatal(err)
}
defer func() {
if err := consumer.Close(); err != nil {
log.Fatal(err)
}
}()
partitions, err := consumer.Partitions(topic)
if err != nil {
log.Fatal(err)
}
for _, partition := range partitions {
go func(partition int32) {
partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
if err != nil {
log.Printf("Failed to start consumer for partition %d: %s\n", partition, err.Error())
return
}
defer func() {
if err := partitionConsumer.Close(); err != nil {
log.Printf("Error closing partition consumer for partition %d: %s\n", partition, err.Error())
}
}()
for message := range partitionConsumer.Messages() {
fmt.Printf("Consumed message: topic=%s partition=%d offset=%d key=%s value=%s\n",
message.Topic, message.Partition, message.Offset, string(message.Key), string(message.Value))
}
}(partition)
}
select {}
}
在上面的代码中，我们使用go关键字在每个分区上启动一个独立的消费者协程，并通过循环遍历所有分区来添加多个消费者。这些消费者将加入到具有相同GroupID的消费组中进行消息消费。
请注意，在实际应用中，你可能需要根据自己的需求对错误处理、关闭和其他配置进行适当的修改。
内容由零声教学AI助手提供，问题来源于学员提问
