Go语言kafka生产消费消息实例搬砖
作者:爱吃红薯粉
这篇文章主要为大家介绍了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的消息
以上就是Go语言kafka生产消费消息实例搬砖的详细内容,更多关于Go语言kafka生产消费消息的资料请关注脚本之家其它相关文章!