当前位置：首页 > news >正文

go语言并发实战——日志收集系统(三) 利用sarama包连接KafKa实现消息的生产与消费

news 2025/8/22 8:13:49

环境的搭建

Kafka以及相关组件的下载

我们要实现今天的内容，不可避免的要进行对开发环境的配置，Kafka环境的配置比较繁琐，需要配置JDK,Scala,ZoopKeeper和Kafka，这里我们不做赘述，如果大家不知道如何配置环境，这里我们个大家找了一篇博文供大家参考：
Windows下安装Kafka（图文记录详细步骤）

sarama包的安装

今天我们所时机的内容需要用到go语言的第三方包sarama，由于1.19版本后添加了ztcd压缩算法，需要用到cgo，这里我们为了方便考虑选择下载sarama v1.19.0，所以这里我们不能直接使用go get'命令来安装第三方包，我们要使用/go mod文件来实现,下面是主要步骤：

在项目中创建文件夹（博主的是Kafkademo）
打开终端，输入go mod init，进行go.mod文件的初始化：
我们在.mod文件内指定第三方包及其版本：

module Kafkademorequire (github.com/Shopify/sarama v1.19
)go 1.21.6

其实这是已经可以使用命令go mod tidy了,但是博主在做的时候发现，这样会直接清除掉.mod文件里面的内容，所以建议先创建一个producer文件，在文件里面写：

package mainimport ("fmt""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.WaitForAll                                
}

这时候再打开终端输入go mod tidy
在这里插入图片描述
等待命令运行完毕，打开.mod文件，看到如下内容就OK了：

利用sarama向Kafka发送消息（消息的生产）

代码

package mainimport ("fmt""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()                              //创建config实例config.Producer.RequiredAcks = sarama.WaitForAll          //发送完数据需要leader和follow都确认config.Producer.Partitioner = sarama.NewRandomPartitioner //创建随机分区config.Producer.Return.Successes = true                   //成功交付的消息将在success channel返回//创建信息msg := &sarama.ProducerMessage{}msg.Topic = "web.log"msg.Value = sarama.StringEncoder("this is a test log")//连接KafKaclient, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)if err != nil {fmt.Println("producer closed, err:", err)return}defer client.Close()//发送消息pid, offset, err := client.SendMessage(msg)if err != nil {fmt.Println("send msg failed,err:", err)return}fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

运行过程

首先我们打开终端开起ZooKepper服务
```
zkServer
```
然后再Kafka所在文件夹下输入命令运行Kafka：

.\bin\windows\kafka-server-start.bat .\config\server.properties

在这里插入图片描述
最后运行程序即可，输出结果为：

补充：消息的消费

代码

package mainimport ("fmt""github.com/Shopify/sarama""time"
)func main() {customer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)if err != nil {fmt.Println("failed init customer,err:", err)return}partitionlist, err := customer.Partitions("web.log-0") //获取topic的所有分区if err != nil {fmt.Println("failed get partition list,err:", err)return}fmt.Println("partitions:", partitionlist)for partition := range partitionlist { // 遍历所有分区//根据消费者对象创建一个分区对象pc, err := customer.ConsumePartition("web.log", int32(partition), sarama.OffsetNewest)if err != nil {fmt.Println("failed get partition consumer,err:", err)return}defer pc.Close() // 移动到这里go func(consumer sarama.PartitionConsumer) {defer pc.AsyncClose() // 移除这行，因为已经在循环结束时关闭了for msg := range pc.Messages() {fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)}}(pc)time.Sleep(time.Second * 10)}
}