Vergleiche für Verbrauchergruppen in Golang Kafka 10 erhalten
Mit der Einführung der Unterstützung von Verbrauchergruppen in Golangs Kafka-Bibliothek (Sarama) in Kafka 10, Entwickler haben jetzt Zugriff auf verbrauchergruppenbezogene Funktionen, ohne auf externe Bibliotheken angewiesen zu sein. Eine häufige Aufgabe ist das Abrufen des aktuellen Nachrichtenoffsets, das von einer Verbrauchergruppe verarbeitet wird. Bisher war hierfür der Einsatz von Zookeeper-basierten Lösungen wie Kazoo-go erforderlich. Mit Sarama-cluster kann dies nun durch den folgenden Code erreicht werden:
<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>Das obige ist der detaillierte Inhalt von**Wie erhält man Verbrauchergruppenausgleiche in Golang Kafka 10?**. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
