go語言中kafka go語言中間件

一次golang sarama kafka內存占用大的排查經歷

環境：

成都創新互聯公司是一家專業提供梅里斯企業網站建設,專注與網站建設、成都網站建設、H5技術、小程序制作等業務。10年已為梅里斯眾多企業、政府機構等服務。創新互聯專業網站設計公司優惠進行中。

現象：golang微服務內存占用超過1G，查看日志發現大量kafka相關錯誤日志，繼而查看kafka集群，其中一個kafka節點容器掛掉了。

疑問 為什么kafka集群只有一個broker掛了，客戶端就大量報錯呢

通過beego admin頁面獲取 mem-1.memprof

可以看到調用棧為 withRecover backgroundMetadataUpdataer refreshMeaatdata RefreshMetada tryRefreshMetadata ...

sarama-cluster: NewClient

為什么kafka集群只有一個broker，但是NewClient確失敗了？

在kafka容器里查看topic, 發現Replicas和Isr只有一個，找到kafka官方配置說明，自動生成的topic需要配置default.replication.factor這個參數，才會生成3副本。

docker 配置 kafka+zookeeper，golang接入示例

配置zookeeper 使用kafka/bin/下自帶的zk

運行 報錯 卒。配置低了

docker-compose.yml

報錯

換云搬瓦工的機器試一下

但是docker ps -a 發現只有zookeeper啟動了，kafka失敗, 檢查日志 發現kafka運行需要java環境，而且對內存有要求，搬瓦工的vps不足夠

因此修改docker-compose.yml 加入以下

stop 再啟動

完美

測試

進入容器

查看已經建好的topic （docker-compose.yml）

發送消息

接收消息

接下來是golang接入kafka了

運行

Golang kafka簡述和操作（sarama同步異步和消費組）

一、Kafka簡述

1. 為什么需要用到消息隊列

異步：對比以前的串行同步方式來說，可以在同一時間做更多的事情，提高效率；

解耦：在耦合太高的場景，多個任務要對同一個數據進行操作消費的時候，會導致一個任務的處理因為另一個任務對數據的操作變得及其復雜。

緩沖：當遇到突發大流量的時候，消息隊列可以先把所有消息有序保存起來，避免直接作用于系統主體，系統主題始終以一個平穩的速率去消費這些消息。

2.為什么選擇kafka呢？

這沒有絕對的好壞，看個人需求來選擇，我這里就抄了一段他人總結的的優缺點，可見原文

kafka的優點：

1.支持多個生產者和消費者2.支持broker的橫向拓展3.副本集機制，實現數據冗余，保證數據不丟失4.通過topic將數據進行分類5.通過分批發送壓縮數據的方式，減少數據傳輸開銷，提高吞高量6.支持多種模式的消息7.基于磁盤實現數據的持久化8.高性能的處理信息，在大數據的情況下，可以保證亞秒級的消息延遲9.一個消費者可以支持多種topic的消息10.對CPU和內存的消耗比較小11.對網絡開銷也比較小12.支持跨數據中心的數據復制13.支持鏡像集群

kafka的缺點：

1.由于是批量發送，所以數據達不到真正的實時2.對于mqtt協議不支持3.不支持物聯網傳感數據直接接入4.只能支持統一分區內消息有序，無法實現全局消息有序5.監控不完善，需要安裝插件6.需要配合zookeeper進行元數據管理7.會丟失數據，并且不支持事務8.可能會重復消費數據，消息會亂序，可用保證一個固定的partition內部的消息是有序的，但是一個topic有多個partition的話，就不能保證有序了，需要zookeeper的支持，topic一般需要人工創建，部署和維護一般都比mq高

3. Golang 操作kafka

3.1. kafka的環境

網上有很多搭建kafka環境教程，這里就不再搭建，就展示一下kafka的環境，在kubernetes上進行的搭建，有需要的私我，可以發yaml文件

3.2. 第三方庫

github.com/Shopify/sarama // kafka主要的庫*github.com/bsm/sarama-cluster // kafka消費組

3.3. 消費者

單個消費者

funcconsumer(){varwg sync.WaitGroup? consumer, err := sarama.NewConsumer([]string{"172.20.3.13:30901"},nil)iferr !=nil{? ? ? fmt.Println("Failed to start consumer: %s", err)return}? partitionList, err := consumer.Partitions("test0")//獲得該topic所有的分區iferr !=nil{? ? ? fmt.Println("Failed to get the list of partition:, ", err)return}forpartition :=rangepartitionList {? ? ? pc, err := consumer.ConsumePartition("test0",int32(partition), sarama.OffsetNewest)iferr !=nil{? ? ? ? fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)return}? ? ? wg.Add(1)gofunc(sarama.PartitionConsumer){//為每個分區開一個go協程去取值formsg :=rangepc.Messages() {//阻塞直到有值發送過來，然后再繼續等待fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))? ? ? ? }deferpc.AsyncClose()? ? ? ? wg.Done()? ? ? }(pc)? }? wg.Wait()}funcmain(){? consumer()}

消費組

funcconsumerCluster(){? groupID :="group-1"config := cluster.NewConfig()? config.Group.Return.Notifications =trueconfig.Consumer.Offsets.CommitInterval =1* time.Second? config.Consumer.Offsets.Initial = sarama.OffsetNewest//初始從最新的offset開始c, err := cluster.NewConsumer(strings.Split("172.20.3.13:30901",","),groupID, strings.Split("test0",","), config)iferr !=nil{? ? ? glog.Errorf("Failed open consumer: %v", err)return}deferc.Close()gofunc(c *cluster.Consumer){? ? ? errors := c.Errors()? ? ? noti := c.Notifications()for{select{caseerr := -errors:? ? ? ? ? ? glog.Errorln(err)case-noti:? ? ? ? }? ? ? }? }(c)formsg :=rangec.Messages() {? ? ? fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))? ? ? c.MarkOffset(msg,"")//MarkOffset 并不是實時寫入kafka，有可能在程序crash時丟掉未提交的offset}}funcmain(){goconsumerCluster()}

3.4. 生產者

同步生產者

packagemainimport("fmt""github.com/Shopify/sarama")funcmain(){? config := sarama.NewConfig()? config.Producer.RequiredAcks = sarama.WaitForAll//賦值為-1：這意味著producer在follower副本確認接收到數據后才算一次發送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//寫到隨機分區中，默認設置8個分區config.Producer.Return.Successes =truemsg := sarama.ProducerMessage{}? msg.Topic =`test0`msg.Value = sarama.StringEncoder("Hello World!")? client, err := sarama.NewSyncProducer([]string{"172.20.3.13:30901"}, config)iferr !=nil{? ? ? fmt.Println("producer close err, ", err)return}deferclient.Close()? pid, offset, err := client.SendMessage(msg)iferr !=nil{? ? ? fmt.Println("send message failed, ", err)return}? fmt.Printf("分區ID:%v, offset:%v \n", pid, offset)}

異步生產者

funcasyncProducer(){? config := sarama.NewConfig()? config.Producer.Return.Successes =true//必須有這個選項config.Producer.Timeout =5* time.Second? p, err := sarama.NewAsyncProducer(strings.Split("172.20.3.13:30901",","), config)deferp.Close()iferr !=nil{return}//這個部分一定要寫，不然通道會被堵塞gofunc(p sarama.AsyncProducer){? ? ? errors := p.Errors()? ? ? success := p.Successes()for{select{caseerr := -errors:iferr !=nil{? ? ? ? ? ? ? glog.Errorln(err)? ? ? ? ? ? }case-success:? ? ? ? }? ? ? }? }(p)for{? ? ? v :="async: "+ strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))? ? ? fmt.Fprintln(os.Stdout, v)? ? ? msg := sarama.ProducerMessage{? ? ? ? Topic: topics,? ? ? ? Value: sarama.ByteEncoder(v),? ? ? }? ? ? p.Input() - msg? ? ? time.Sleep(time.Second *1)? }}funcmain(){goasyncProducer()select{? ? ? }}

3.5. 結果展示-

同步生產打印：

分區ID:0,offset:90

消費打印：

Partition:0,Offset:90,key:,value:Hello World!

異步生產打印：

async:7272async:7616async:998

消費打印：

Partition:0,Offset:91,key:,value:async:7272Partition:0,Offset:92,key:,value:async:7616Partition:0,Offset:93,key:,value:async:998

聊聊golang的zap的ZapKafkaWriter

本文主要研究一下golang的zap的ZapKafkaWriter

WriteSyncer內嵌了io.Writer接口，定義了Sync方法；Sink接口內嵌了zapcore.WriteSyncer及io.Closer接口；ZapKafkaWriter實現Sink接口及zapcore.WriteSyncer接口，其Write方法直接將data通過kafka發送出去。

新聞標題：go語言中kafka go語言中間件
分享網址：http://m.kartarina.com/article20/dogeeco.html

成都網站建設公司_創新互聯，為您提供微信公眾號、App設計、網頁設計公司、電子商務、響應式網站、網站制作

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯