首页  >  文章  >  后端开发  >  **如何在 Golang Kafka 10 中获取消费者组偏移量?**

**如何在 Golang Kafka 10 中获取消费者组偏移量?**

DDD
DDD原创
2024-10-25 10:51:02871浏览

**How to Get Consumer Group Offsets in Golang Kafka 10?**

在 Golang Kafka 10 中获取消费者组偏移

随着 Kafka 10 中 Golang 的 Kafka 库(Sarama)引入消费者组支持,开发人员现在可以访问与消费者组相关的功能,而无需依赖外部库。一项常见任务是检索消费者组正在处理的当前消息偏移量。以前,这需要使用基于 Zookeeper 的解决方案,例如 Kazoo-go。现在,有了 Sarama-cluster,这可以通过以下代码来实现:

<code class="go">package main

import (
    "context"
    "log"
    "strings"

    "github.com/Shopify/sarama"
)

func main() {
    groupName := "testgrp"
    topic := "topic_name"
    offset, err := GetCGOffset(context.Background(), "localhost:9092", groupName, topic)
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Consumer group %s offset for topic %s is: %d", groupName, topic, offset)
}

type gcInfo struct {
    offset int64
}

func (g *gcInfo) Setup(sarama.ConsumerGroupSession) error {
    return nil
}

func (g *gcInfo) Cleanup(sarama.ConsumerGroupSession) error {
    return nil
}

func (g *gcInfo) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    g.offset = claim.InitialOffset()
    return nil
}

func GetCGOffset(ctx context.Context, brokers, groupName, topic string) (int64, error) {
    config := sarama.NewConfig()
    config.Consumer.Offsets.AutoCommit.Enable = false // we're not going to update the consumer group offsets
    client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), groupName, config)
    if err != nil {
        return 0, err
    }
    info := gcInfo{}
    if err := client.Consume(ctx, []string{topic}, &info); err != nil {
        return 0, err
    }
    return info.offset, nil
}</code>

以上是**如何在 Golang Kafka 10 中获取消费者组偏移量?**的详细内容。更多信息请关注PHP中文网其他相关文章!

声明:
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn