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

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

2022-10-17 11:05爱吃红薯粉 Golang

这篇文章主要为大家介绍了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)

									           }

									      }

									   }

									}