Go语言kafka生产消费消息实例搬砖

kafka go库

kafka go客户端官方目前没有提供，但在github有2个非常流行的库

星星较多，网上案例也多 https://github.com/Shopify/sarama

confluent官网提供的库 https://github.com/confluentinc/confluent-kafka-go

这里使用sarama，因为星星多，案例多，方便快速上手

注意

如果kafka版本在2.2以下，需要在go.mod里面将sarama版本改为github.com/Shopify/sarama v1.24.1

这是因为sarama只提供最新2个版本+2个月的兼容保证，所以使用低版本kafka是需要避坑

使用非集群集群生产者时，需要自行去创建topic，如果使用集群生产者，集群会自动创建

例子

package main
import (
  "fmt"
  "github.com/Shopify/sarama"
  cluster "github.com/bsm/sarama-cluster"
  "time"
)
var (
  Consumer *cluster.Consumer
  producer sarama.SyncProducer
  brokers = []string{"ip1:9092","ip2:9092","ip3:9092"}
  topic = "testGo"
  groupId = "testGo_test1"
)
func initProducer() {
  var err error
  config := sarama.NewConfig()
  config.Producer.RequiredAcks = sarama.WaitForLocal
  config.Producer.Retry.Max = 3
  config.Producer.Return.Successes = true
  brokers := brokers
  producer, err = sarama.NewSyncProducer(brokers,config)
  if err != nil {
     fmt.Printf("生产者初始化失败 -> ％v \n", err)
     panic(err)
  }
  fmt.Println("生产者初始化成功。")
}
func initConsumer()  {
  var err error
  config := cluster.NewConfig()
  config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
  config.Consumer.Offsets.Initial = sarama.OffsetNewest
  Consumer, err = cluster.NewConsumer(brokers,groupId,[]string{topic},config)
  if err != nil {
     fmt.Printf("消费者初始化失败 -> ％v \n", err)
     panic(err.Error())
  }
  if Consumer == nil {
     panic(fmt.Sprintf("消费者为空. kafka info -> {brokers:％v, topic: ％v, group: ％v}", brokers, topic, groupId))
  }
  fmt.Printf("消费者初始化成功, consumer -> ％v, topic -> ％v, ", Consumer, topic)
}
func main() {
  initProducer()
  initConsumer()
  //生产消息
  for i := 1;i < 100; i ++ {
     pid, offset, err := producer.SendMessage(&sarama.ProducerMessage{
        Topic:     topic,
        Key:       sarama.StringEncoder(i),
        Value:     sarama.ByteEncoder("this is test message."),
     })
     if err != nil {
        fmt.Println("发送消息失败, err:", err)
        return
     }
     fmt.Printf("offset: ％v\n", offset)
  }
  time.Sleep(2 * time.Second)
  //消费消息
  for {
      select {
      case msg, ok := <-Consumer.Messages():
          if ok {
             fmt.Printf("kafka msg: ％s \n", msg.Value)
          }
     }
  }
}

运行结果如下

该demo流程如下

• 引入单机sarama库和集群sarama库

• 定义连接变量

• 使用单sarama库实例化一个生产者

• 使用集群sarama库实例化一个消费者

• 循环100次发送100条消息

• 使用sarama自带的生产消息构造器设置消息内容

• 使用for让进程一直监听来自kafka的消息

来源：https://juejin.cn/post/7033565443084255263