Sarama を使用して Golang Kafka 10 のコンシューマ グループの現在のメッセージ オフセットを取得する方法

Golang Kafka 10 を使用したコンシューマ グループ オフセットの取得

従来、Kafka を使用して Golang のコンシューマ グループ機能を管理するには、外部ライブラリが採用されていました。ただし、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 don't want to change 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>

以上がSarama を使用して Golang Kafka 10 のコンシューマ グループの現在のメッセージ オフセットを取得する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。