登录
首页 >  Golang >  Go教程

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

来源:脚本之家

时间:2022-12-29 10:51:32 395浏览 收藏

IT行业相对于一般传统行业,发展更新速度更快,一旦停止了学习,很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习,精进自己的技术,尤其是初学者。今天golang学习网给大家整理了《Go语言kafka生产消费消息实例搬砖》,聊聊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 

运行结果如下

该demo流程如下

  • 引入单机sarama库和集群sarama库
  • 定义连接变量
  • 使用单sarama库实例化一个生产者
  • 使用集群sarama库实例化一个消费者
  • 循环100次发送100条消息
  • 使用sarama自带的生产消息构造器设置消息内容
  • 使用for让进程一直监听来自kafka的消息

理论要掌握,实操不能落!以上关于《Go语言kafka生产消费消息实例搬砖》的详细介绍,大家都掌握了吧!如果想要继续提升自己的能力,那么就来关注golang学习网公众号吧!

声明:本文转载于:脚本之家 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>