go语言操作kafka
Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据,具有高性能、持久化、多副本备份、横向扩展等特点。本文介绍了如何使用Go语言发送和接收kafka消息。
Sarama
Go语言中连接kafka使用第三方库:github.com/Shopify/sarama。
下载及安装
go get github.com/Shopify/Sarama
注意事项
Sarama
v1.20之后的版本加入了zstd
压缩算法,需要用到cgo,在Windows平台编译时会提示类似如下错误:
# github.com/DataDog/zstd
exec: "gcc":executable file not found in %PATH%
所以在Windows平台请使用v1.19版本的Sarama。
连接kafka发送消息
package main
import (
"fmt"
"github.com/Shopify/Sarama"
)
// 基于Sarama第三方库开发的kafka client
func main() {
config := Sarama.NewConfig()
config.Producer.requiredAcks = Sarama.WaitForAll // 发送完数据需要leader和follow都确认
config.Producer.Partitioner = Sarama.NewRandomPartitioner // 新选出一个partition
config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
// 构造一个消息
msg := &Sarama.ProducerMessage{}
msg.Topic = "web_log"
msg.Value = Sarama.StringEncoder("this is a test log")
// 连接kafka
client, err := Sarama.NewSyncProducer([]string{"192.168.1.7: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)
}
连接kafka消费消息
package main
import (
"fmt"
"github.com/Shopify/Sarama"
)
// kafka consumer
func main() {
consumer, err := Sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
if err != nil {
fmt.Printf("fail to start consumer, err:%v\n", err)
return
}
partitionList, err := consumer.Partitions("web_log") // 根据topic取到所有的分区
if err != nil {
fmt.Printf("fail to get list of partition:err%v\n", err)
return
}
fmt.Println(partitionList)
for partition := range partitionList { // 遍历所有的分区
// 针对每个分区创建一个对应的分区消费者
pc, err := consumer.ConsumePartition("web_log", int32(partition), Sarama.OffsetNewest)
if err != nil {
fmt.Printf("Failed to start consumer for partition %d,err:%v\n", partition, err)
return
}
defer pc.Asyncclose()
// 异步从每个分区消费信息
go func(Sarama.PartitionConsumer) {
for msg := range pc.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
}
}(pc)
}
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。