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 サイトの他の関連記事を参照してください。