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，这个是尝试了很多才找到的
  2. 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

  比较冗余和复杂，还有考虑到当时的时间问题，没有进行具体的测试