Golang Go语言中折腾 kafka

发布于 1周前 作者 sinazl 来自 Go语言

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

3 回复

嗯,所以你写了这么长,到底表达了什么呢?

更多关于Golang Go语言中折腾 kafka的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html


这是把 V2 当作自己的笔记本的意思

在Golang中集成和使用Kafka是一项常见且强大的任务,它能够帮助你构建高效、可扩展的实时数据处理系统。Kafka作为一个分布式流处理平台,在日志收集、实时流处理等方面有着广泛的应用。

在Golang中操作Kafka,你通常会用到一些开源的客户端库,比如Shopify/saramaconfluent-kafka-go。这些库提供了丰富的API,让你能够轻松地与Kafka集群进行交互,包括生产消息、消费消息、管理主题等。

使用sarama库时,你需要先创建一个Kafka客户端,然后通过这个客户端连接到Kafka集群。之后,你可以创建生产者来发送消息到指定的主题,或者创建消费者来从主题中读取消息。记得要处理好连接管理、错误处理以及消息的序列化与反序列化。

对于实时性要求较高的应用,confluent-kafka-go可能是一个更好的选择,因为它提供了更底层的C语言绑定,能够带来更高的性能。不过,这也意味着你需要更深入地了解Kafka的内部机制。

在整合Kafka到你的Golang应用时,还需要注意一些常见的陷阱,比如消息丢失、重复消费、消费滞后等问题。这些问题通常可以通过合理配置Kafka和消费逻辑来解决。

总的来说,虽然Kafka的集成和使用可能看起来有些复杂,但只要你理解了它的基本原理和API,就能够充分利用它的强大功能来构建出高效、稳定的实时数据处理系统。

回到顶部