我正在使用 Kafka 10.0 和 https://github.com/Shopify/sarama . 我正在尝试获取消费者处理的最新消息的偏移量。
为此,我找到了方法 NewOffsetManagerFromClient(group string, client Client)需要组名。
如何获取消费者组名称?
offsets := make(map[int32]int64)
config := sarama.NewConfig()
config.Consumer.Offsets.CommitInterval = 200 * time.Millisecond
config.Version = sarama.V0_10_0_0
// config.Consumer.Offsets.Initial = sarama.OffsetNewest
cli, _ := sarama.NewClient(kafkaHost, config)
defer cli.Close()
offsetManager, _ := sarama.NewOffsetManagerFromClient(group, cli)
for _, partition := range partitions {
partitionOffsetManager, _ := offsetManager.ManagePartition(topic, partition)
offset, _ := partitionOffsetManager.NextOffset()
offsets[partition] = offset
}
return offsets
我创建了一个消费者
consumer := sarama.NewConsumer(connections, config)
但我不知道如何创建消费者组并获取其组名。
最佳答案
您正在尝试创建自己的偏移量管理器以查找当前偏移量:
offsetManager, _ := sarama.NewOffsetManagerFromClient(group, cli)
类似地,消费您的主题消息的消费者必须使用相同的偏移量管理器,并且他们将使用特定的组 ID。使用该组 ID。
关于go - 如何在 Golang Kafka 10 中获取 GroupID?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41984572/