如何在 Golang 中创建 kafka 消费组？答案

【问题标题】：How to create a kafka consumer group in Golang?如何在 Golang 中创建 kafka 消费组？
【发布时间】：2017-06-18 14:34:05
【问题描述】：

一个可用的库是sarama（或它的扩展sarama-cluster），但没有提供消费者组示例，sarama 和sarama-cluster 都没有。

我不懂 API。我可以举一个为主题创建消费者组的示例吗？

【问题讨论】：

标签： go apache-kafka sarama

【解决方案1】：

无需使用 sarama-cluster 库。不推荐用于 apache kafka 集成。 Sarama原始库本身提供了一种使用消费者组连接到kafka集群的方法。

我们需要创建客户端，然后初始化消费者组，我们在其中创建声明并等待消息通道接收消息。

初始化客户端：-

kfversion, err := sarama.ParseKafkaVersion(kafkaVersion) // kafkaVersion is the version of kafka server like 0.11.0.2
if err != nil {
    log.Println(err)
}

config := sarama.NewConfig()
config.Version = kfversion
config.Consumer.Return.Errors = true

// Start with a client
client, err := sarama.NewClient([]string{brokerAddr}, config)
if err != nil {
    log.Println(err)
}
defer func() { _ = client.Close() }()

连接到消费者组：-

// Start a new consumer group
group, err := sarama.NewConsumerGroupFromClient(consumer_group, client)
if err != nil {
    log.Println(err)
}
defer func() { _ = group.Close() }()

从主题分区开始消费消息：-

// Iterate over consumer sessions.
ctx := context.Background()
for {
    topics := []string{topicName}
    handler := &Message{}
    err := group.Consume(ctx, topics, handler)
    if err != nil {
        log.Println(err)
    }
}

最后一部分是等待消息通道消费消息。我们需要实现所有的功能（三个）来实现ConsumerGroupHandler接口。

func (exampleConsumerGroupHandler) Setup(_ ConsumerGroupSession) error   { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess ConsumerGroupSession, claim ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        fmt.Printf("Message topic:%q partition:%d offset:%d\n", msg.Topic, msg.Partition, msg.Offset)
        sess.MarkMessage(msg, "")
    }
    return nil
}

有关使用 golang 的 kafka 的更多信息，请查看 sarama 库。

【讨论】：

除非卡在 kafka 0.10.1，否则还是需要使用 sarama-cluster
如果我在 Kubernetes Deployment 的 Pod 中部署这个程序，我可以安全地扩展吗？
是的@LiZiyan，您可以将生产者扩展到集群中的多个 pod

【解决方案2】：

消费者组由集群消费者“构造函数”的第二个参数指定。这是一个非常基本的草图：

import (
    "github.com/Shopify/sarama"
    "github.com/bsm/sarama-cluster"
)

conf := cluster.NewConfig()
// add config values

brokers := []string{"kafka-1:9092", "kafka-2:9092"}
group := "Your-Consumer-Group"
topics := []string{"topicName"}
consumer := cluster.NewConsumer(broker, group, topics, conf)

因此您将拥有一个属于指定消费者组的消费者。

【讨论】：