golang-封装kafka消费端

在 Golang 中使用 Kafka 消费端需要使用第三方库 sarama，它是一个 Kafka 客户端库，提供了 Kafka 生产者和消费者的实现，支持高级别消费者和低级别消费者。

下面是一个简单的封装 Kafka 消费端的示例：

package kafka

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

type Consumer struct {
    brokers []string
    topic   string
    group   string
    config  *sarama.Config
    handler func(message []byte) error
}

func NewConsumer(brokers []string, topic string, group string, handler func(message []byte) error) *Consumer {
    config := sarama.NewConfig()
    config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
    config.Consumer.Offsets.Initial = sarama.OffsetOldest
    return &Consumer{
        brokers: brokers,
        topic:   topic,
        group:   group,
        config:  config,
        handler: handler,
    }
}

func (c *Consumer) Consume() error {
    consumer, err := sarama.NewConsumerGroup(c.brokers, c.group, c.config)
    if err != nil {
        return fmt.Errorf("Error creating consumer group: %v", err)
    }
    defer consumer.Close()

    handler := consumerHandler{handler: c.handler}

    ctx := sarama.ConsumerGroupSessionContext(nil)
    for {
        err = consumer.Consume(ctx, []string{c.topic}, handler)
        if err != nil {
            return fmt.Errorf("Error consuming topic %s: %v", c.topic, err)
        }
    }

    return nil
}

type consumerHandler struct {
    handler func(message []byte) error
}

func (h consumerHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h consumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (h consumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for message := range claim.Messages() {
        err := h.handler(message.Value)
        if err != nil {
            return err
        }
        session.MarkMessage(message, "")
    }
    return nil
}

在这个示例中，我们定义了一个名为 Consumer 的结构体，它包含了 Kafka 的配置信息、topic、group 和 message 处理函数。在 NewConsumer 函数中，我们创建了一个新的 Consumer，并返回其指针。在 Consume 函数中，我们创建了一个新的 ConsumerGroup，并使用 sarama.ConsumerGroupSessionContext 创建了一个新的上下文。然后我们使用 sarama.ConsumerGroup 的 Consume 方法来消费消息。消费消息时，我们将消息传递给 handler 函数，并使用 session.MarkMessage 将消息标记为已处理。消费结束后，我们返回 nil。

这是一个简单的封装 Kafka 消费端的示例，你可以根据自己的需求进行修改和扩展。