在Golang Kafka 10 中取得消費者群組偏移
隨著Kafka 10 中Golang 的Kafka 庫(Sarama)引入消費者庫(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中文網其他相關文章!