如何使用 Kafka 10 檢索 Go 中的消費者群組偏移量？

使用Kafka 10 檢索Go 中的消費者群組偏移量

隨著Kafka 10 的發布，Go Kafka 庫(sarama)現在提供消費者無需依賴外部庫即可對功能進行分組。這就提出瞭如何檢索消費者群組正在處理的當前訊息偏移量的問題。

解決方案

要取得消費者群組偏移量，請依照下列步驟操作：

1. 實現消費者群組資訊結構：

   <code class="go">type gcInfo struct {
       offset int64
   }</code>

2. 建立消費者群組資訊處理程序：

   <code class="go">func (g *gcInfo) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
       g.offset = claim.InitialOffset()
       return nil
   }</code>

3. 設定並建立消費者群組：

   <code class="go">config := sarama.NewConfig()
   config.Consumer.Offsets.AutoCommit.Enable = false
   client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), groupName, config)</code>

4. 消費群組內訊息：

   <code class="go">info := gcInfo{}
   if err := client.Consume(ctx, []string{topic}, &amp;info); err != nil {
       return 0, err
   }</code>

5. 檢索偏移量：

   <code class="go">return info.offset, nil</code>

以此實現後，您可以在任何給定時間檢索特定分區和主題的消費者組偏移量。

以上是如何使用 Kafka 10 檢索 Go 中的消費者群組偏移量？的詳細內容。更多資訊請關注PHP中文網其他相關文章！