Golang中常用的kafka库

本文将介绍目前主流的三种第三方kafka库，并结合实际使用中遇到的问题，给出实际的解决方案，本文只涉及kafka数据的消费，且其中对所有库的使用仅为测试过的简单代码，仅记录自己踩坑过程。
问题：最开始使用的是sarama-cluster库，git地址为链接: github.com/bsm/sarama-cluster，但是这个库在使用过程中，存在从头消费kafka数据问题，实际消费场景是实时消费kafka数据，其中有关kafka的配置如下：

package main

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

func main() {
    clusterConfig := cluster.NewConfig()
    clusterConfig.Consumer.Return.Errors = true
    clusterConfig.Group.Return.Notifications = true
    clusterConfig.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
    clusterConfig.Version = sarama.V0_11_0_0
    clusterConfig.Consumer.Offsets.CommitInterval = 1 * time.Second
    clusterConfig.Consumer.Offsets.Initial = sarama.OffsetNewest

    // TODO: 添加你的业务逻辑
}

sarama-cluster已经废弃了，没有人维护了，所以遇到的问题不易找到解决方案，故考虑换库。
注意：在实际使用中，需要考虑一个现实问题，生产数据的kafka集群和消费kafka的集群在两台服务器，且两台服务器之间并没有进行时钟同步，故基于此情况下考虑去消费实时数据

1. kafka-go

首先考虑比较轻量的kafka-go库，github的地址为链接: github.com/segmentio/kafka-go，具体的代码示例如下：

package main

import (
   "time"
   "github.com/segmentio/kafka-go"
)

func main() {
   brokers := "192.168.13.22:9092"
   groupID := "test"
   topic := "example"
   consumer := kafka.NewReader(kafka.ReaderConfig{
       Brokers:     brokers,
       GroupID:     groupID,
       Topic:       topic,
       MinBytes:    10e3, // 10KB
       MaxBytes:    10e6, // 10MB
       StartOffset: kafka.LastOffset, // 这个很关键，决定了是否是从最新的位置消费数据
   })
   // 消费数据
   for {
       msg, err := consumer.ReadMessage(context.Background())
       if err != nil {
           fmt.Printf("kafka消费异常,err:%v\n", err)
           continue
       }
       fmt.Printf("Message on partition%d, offset %d, topic %s, msg:%s, time[%v]\n", msg.Partition,
           msg.Offset, msg.Topic, string(msg.Value), msg.Time)
       // TODO: 添加你的业务逻辑
   }
}

踩坑点：经过多次尝试和实际代码调试，kafka-go库中需要注意以下几点：

• broker是slice，故可以设置多个，但是topic是string类型只能是一个，所以如果有多个topic需要消费，只能采取使用NewReader去创建多个kafka消费者实例
• 关于group id的设置，真的是踩坑无数，如果group id 设置为空的话，只会消费partition为0的数据。group id相同可能会接着上次未消费的数据进行消费而不是实时消费，这个结论没有验证，但是为了保险起见，多个consumer实例设置为不同的group id没有任何问题
• 如果需要实时消费，即消费最新的数据，一定要设置StartOffset字段为kafka.LastOffset，这个是尝试了很多才找到的

1. confluent-kafka-go git上的地址为：github.com/confluentinc/confluent-kafka-go/kafka 创建kafkaConsumer实例及消费数据如下：

func NewConFluentConsumer(broker string, groupId string, topics []string) (*kafka.Consumer, error) {
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": broker,
        "group.id":          groupId,
        "auto.offset.reset": "latest",
    })
    if err != nil {
        fmt.Printf("new confluent consumer failed, err[%v]\n", err)
        return nil, err
    }

    err = c.SubscribeTopics(topics, nil)
    if err != nil {
        fmt.Printf("subscribe topic[%v] failed, err[%v]\n", topics, err)
        return nil, err
    }

    fmt.Printf"new confluent consumer success, broker[%s], groupId[%s], topic[%v]\n", broker, groupId, topics)
    return c, nil
}

func ConsumerMessageWorker(consumer *kafka.Consumer) {
    if consumer == nil {
        return
    }
    for {
        msg, err := consumer.ReadMessage(-1)
        if err == nil {
            fmt.Printf("Message on partition%d, topic %s, msg:%s\n", msg.TopicPartition.Partition,*msg.TopicPartition.Topic, string(msg.Value))
        // TODO 增加消费到的消息处理逻辑
        } else if !err.(kafka.Error).IsTimeout() {
            // 没超时，但是报错了
            // The client will automatically try to recover from all errors.
            // Timeout is not considered an error because it is raised by
            // ReadMessage in absence of messages.
            fmt.Printf("Consumer error: %v (%v)\n", err, msg)
        }
    }
}

其中kafka消费配置使用的map参数可以在https://github.com/confluentinc/librdkafka/tree/master/CONFIGURATION.md中找到相关的描述
经测试，是可以支持单个broker，多个topic对数据进行消费，但是考虑到此库使用了c库，故在跨平台方面可能对程序的移植存在兼容性问题，故弃用。
阿里云的相关网站上也有kafka消费的一些示例，可以参考，链接如下：https://help.aliyun.com/zh/sls/user-guide/use-confluent-kafka-go-to-achieve-kafka-consumption
3. sarama
比较冗余和复杂，还有考虑到当时的时间问题，没有进行具体的测试