1

我正在使用 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)

但我不知道如何创建消费者组并获取其组名。

4

2 回答 2

0

我认为您可以使用任何字符串作为 groupId。请查看sarama GoDoc中的示例

// Start a new consumer group
group, err := NewConsumerGroupFromClient("my-group", client)
if err != nil {
    panic(err)
}
defer func() { _ = group.Close() }()

也许你可以给它任何字符串。并且您应该确保其他消费者可以获得相同的 groupId 来加入该组。

于 2019-12-28T17:41:48.967 回答
0

您正在尝试创建自己的偏移管理器来查找当前偏移：

offsetManager, _ := sarama.NewOffsetManagerFromClient(group, cli)

同样，消费主题消息的消费者必须使用相同的偏移管理器，并且他们将使用特定的组 ID。使用该组 ID。

于 2017-02-01T18:10:04.687 回答