如何使用 Sarama 检索 Golang Kafka 10 中的消费者组偏移量？

访问 Golang Kafka 10 中的消费者组偏移

Golang Kafka 库（sarama）与 Kafka 10 的演变引入了本机消费者组功能。这就提出了如何检索消费者组正在处理的当前消息偏移量的问题。

解决方案：

为了解决这个需求，sarama 库提供了一种机制用于访问消费者组偏移量。以下代码片段演示了如何使用 GetCGOffset() 获取偏移量：

<code class="go">import (
    "context"
    "fmt"
    "strings"
    "github.com/Shopify/sarama"
)

func GetCGOffset(ctx context.Context, brokers, groupName, topic string) (int64, error) {
    config := sarama.NewConfig()
    config.Consumer.Offsets.AutoCommit.Enable = false

    client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), groupName, config)
    if err != nil {
        return 0, err
    }
    defer client.Close()

    info := gcInfo{}
    if err := client.Consume(ctx, []string{topic}, &info); err != nil {
        return 0, err
    }
    return info.offset, nil
}</code>

此代码创建一个消费者组并连接到指定的代理。它禁用自动提交以确保偏移量在检索过程中不会更新。 gcInfo 结构体跟踪初始偏移量，该偏移量代表消费者组当前正在处理的消息偏移量。

通过使用此机制，开发人员可以轻松检查和管理其消费者组的偏移量，从而能够更精细地控制消息处理和跟踪。

以上是如何使用 Sarama 检索 Golang Kafka 10 中的消费者组偏移量？的详细内容。更多信息请关注PHP中文网其他相关文章！