Go语言kafka生产消费消息实例搬砖
作者:爱吃红薯粉 时间:2024-06-07 16:06:58
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
标签:Go,kafka,生产消费
0
投稿
猜你喜欢
使用rpclib进行Python网络编程时的注释问题
2022-12-26 23:44:47
Python导入自定义路径的方法
2021-06-30 13:43:28
python调用外部程序的实操步骤
2021-09-11 10:39:37
cropper js基于vue的图片裁剪上传功能的实现代码
2024-05-11 09:10:50
Flask框架debug与配置项的开启与设置详解
2022-04-11 17:05:58
VS CODE 使用SVN插件的方法步骤
2023-05-22 10:39:33
Python通过命令开启http.server服务器的方法
2022-10-08 01:42:41
python中使用矢量化替换循环详解
2023-08-27 00:51:01
JS组件Bootstrap实现图片轮播效果
2024-04-22 13:03:11
python实现mask矩阵示例(根据列表所给元素)
2022-04-19 19:37:16
使用SQL语句快速获取SQL Server数据字典
2009-01-08 16:31:00
python 实现关联规则算法Apriori的示例
2021-05-21 08:45:30
python-opencv 将连续图片写成视频格式的方法
2023-08-25 23:00:06
php 解压zip压缩包内容到指定目录的实例
2024-04-18 09:28:47
mysql5.7.19 安装配置方法图文教程(win10)
2024-01-16 03:52:10
python写入xml文件的方法
2023-01-01 15:08:54
Selenium鼠标与键盘事件常用操作方法示例
2023-04-23 12:39:26
Python3爬虫学习之爬虫利器Beautiful Soup用法分析
2021-04-13 07:01:50
从列表或字典创建Pandas的DataFrame对象的方法
2022-06-12 13:03:45
Mozilla 表达式 __noSuchMethod__
2024-04-18 09:42:21