Go語言kafka生產消費消息實例搬磚

Posted on by

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的消息

以上就是Go語言kafka生產消費消息實例搬磚的詳細內容，更多關於Go語言kafka生產消費消息的資料請關註WalkonNet其它相關文章！

推薦閱讀：