Go Kafka Sarama 消费者组消费代码示例

下面是使用 Sarama 库的 Kafka 消费组消费代码示例：

package main

import (
    "fmt"
    "os"
    "os/signal"
    "github.com/Shopify/sarama"
)

func main() {
    config := sarama.NewConfig()
    config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
    config.Consumer.Offsets.Initial = sarama.OffsetOldest

    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := consumer.Close(); err != nil {
            panic(err)
        }
    }()

    signals := make(chan os.Signal, 1)
    signal.Notify(signals, os.Interrupt)

    partitionConsumer, err := consumer.ConsumePartition("my_topic", 0, sarama.OffsetNewest)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            panic(err)
        }
    }()

    for {
        select {
        case msg := <-partitionConsumer.Messages():
            fmt.Printf("Received message: %s\n", string(msg.Value))
        case <-signals:
            return
        }
    }
}

这个示例中创建了一个 Kafka 消费者，使用 RoundRobin 策略进行消费者组的负载均衡，并从最早的偏移量开始消费。然后，从指定的主题和分区创建了一个分区消费者，并循环消费消息。如果收到中断信号，则退出循环。