Golang Go语言中折腾 kafka
Golang Go语言中折腾 kafka
Apache Kafka
Apache Kafka 是一个分布式发布 - 订阅消息系统和一个强大的队列,可以处理大量的数据,并使您能够将消息从一个端点传递到另一个端点。
- 特点
- 分布系统
- 订阅消息
- 强大队列
- 数据落地到硬盘
安装
- 前置条件:需要 zookeeper。
- 安装 zookeeper
- 官方下载
- 接下即安装( java 生态就是好)
- conf 下 sample 文件改成 zoo.cfg (默认单机模式)
- zookeeper 常用命令
//开启 zookeeper bin/zkServer.sh start //使用客户端连接 bin/zkCli.sh //远程连接 bin/zkCli.sh -server ip:port //停止 zookeeper bin/zkServer.sh stop
- 其他相关 zookeeper 不细讲用到在学习
- 安装 kafaka
- 官方下载
- 接下即安装( java 生态就是好)
- kafaka 常用命令
//开启 kafaka bin/kafka-server-start.sh config/server.properties //创建 Topic bin/kafka-topics.sh --create --zookeeper 192.168.112.20:2181 --replication-factor 1 --partitions 1 --topic test //Topic 展示 bin/kafka-topics.sh --list --zookeeper 192.168.112.20:2181
遇到的问题:下载的是 kafka 的源码包,执行报错 classpath is empty. please build the project first e.g. by running 'gradlew jarall' 应该下载二进制包 下载地址:https://kafka.apache.org/downloads
使用 Go 实现生产消费
- 实现生产者
package main
import (
“bufio”
“fmt”
“github.com/Shopify/sarama”
“os”
“strings”
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
//初始化生产者
producer, err := sarama.NewSyncProducer([]string{“192.168.112.20:9092”}, config)
if err != nil {
panic(err)
}
defer producer.Close()
msg := &sarama.ProducerMessage{
Topic: "testGo",
Partition: int32(-1),
Key: sarama.StringEncoder("key"),
}
var value string
for {
// 生产消息
inputReader := bufio.NewReader(os.Stdin)
value, err = inputReader.ReadString('\n')
if err != nil {
panic(err)
}
value = strings.Replace(value, "\n", "", -1)
msg.Value = sarama.ByteEncoder(value)
//发送消息
paritition, offset, err := producer.SendMessage(msg)
if err != nil {
fmt.Println("Send Message Fail")
}
//输出结果
fmt.Printf("Partion = %d, offset = %d\n", paritition, offset)
}
}
- 实现消费者
package main
import (
“fmt”
“github.com/Shopify/sarama”
“sync”
)
//协成锁 等待所有协成执行完毕退出
var wg sync.WaitGroup
func main() {
//消费者初始化
consumer, err := sarama.NewConsumer([]string{“192.168.112.20:9092”}, nil)
if err != nil {
panic(err)
}
partitionList, err := consumer.Partitions(“testGo”)
if err != nil {
panic(err)
}
for partition := range partitionList {
pc, err := consumer.ConsumePartition(“testGo”, int32(partition), sarama.OffsetNewest)
if err != nil {
panic(err)
}
defer pc.AsyncClose()
//上锁
wg.Add(1)
go func(sarama.PartitionConsumer) {
//解锁
defer wg.Done()
for msg := range pc.Messages() {
//接受消息
fmt.Printf(“Partition:%d, Offset:%d, Key:%s, Value:%s\n”, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
}
}(pc)
//等待全部解锁
wg.Wait()
consumer.Close()
}
}
遇到问题 github.com/Shopify/sarama 包在 windows 环境下需要 gcc,懒得搞直接虚拟机。
总结
利用 go + kafka 实现生产者消费模型非常简单和快速,在日常业务上使用消息队列可以考虑 kafka+go 快速迭代,后续应用上采坑在来分享。
更多关于Golang Go语言中折腾 kafka的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html
嗯,所以你写了这么长,到底表达了什么呢?
更多关于Golang Go语言中折腾 kafka的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html
这是把 V2 当作自己的笔记本的意思
在Golang中集成和使用Kafka是一项常见且强大的任务,它能够帮助你构建高效、可扩展的实时数据处理系统。Kafka作为一个分布式流处理平台,在日志收集、实时流处理等方面有着广泛的应用。
在Golang中操作Kafka,你通常会用到一些开源的客户端库,比如Shopify/sarama
或confluent-kafka-go
。这些库提供了丰富的API,让你能够轻松地与Kafka集群进行交互,包括生产消息、消费消息、管理主题等。
使用sarama
库时,你需要先创建一个Kafka客户端,然后通过这个客户端连接到Kafka集群。之后,你可以创建生产者来发送消息到指定的主题,或者创建消费者来从主题中读取消息。记得要处理好连接管理、错误处理以及消息的序列化与反序列化。
对于实时性要求较高的应用,confluent-kafka-go
可能是一个更好的选择,因为它提供了更底层的C语言绑定,能够带来更高的性能。不过,这也意味着你需要更深入地了解Kafka的内部机制。
在整合Kafka到你的Golang应用时,还需要注意一些常见的陷阱,比如消息丢失、重复消费、消费滞后等问题。这些问题通常可以通过合理配置Kafka和消费逻辑来解决。
总的来说,虽然Kafka的集成和使用可能看起来有些复杂,但只要你理解了它的基本原理和API,就能够充分利用它的强大功能来构建出高效、稳定的实时数据处理系统。