golang的回調(diào)和接口

最近寫了個(gè)kafka的接收消息的功能，需要使用回調(diào)處理收到的消息。

讓客戶滿意是我們工作的目標(biāo)，不斷超越客戶的期望值來自于我們對(duì)這個(gè)行業(yè)的熱愛。我們立志把好的技術(shù)通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領(lǐng)域值得信任、有價(jià)值的長期合作伙伴，公司提供的服務(wù)項(xiàng)目有：主機(jī)域名、雅安服務(wù)器托管、營銷軟件、網(wǎng)站建設(shè)、沁縣網(wǎng)站維護(hù)、網(wǎng)站推廣。

一個(gè)是基本的回調(diào)，一個(gè)是使用接口功能實(shí)現(xiàn)回調(diào)，對(duì)接口是個(gè)很好的學(xué)習(xí)。

1.正常回調(diào)

kafka的接收消息處。收到消息后，使用傳入的Onmessage進(jìn)行處理。

調(diào)用kafka接收消息的單元，并在調(diào)用方寫好回調(diào)

在調(diào)用方實(shí)現(xiàn)回調(diào)需要執(zhí)行的方法

感覺還是使用基本回調(diào)相對(duì)簡單點(diǎn)，接口就當(dāng)學(xué)習(xí)了。

另外跨包的接口的方法要大寫！定位了好久發(fā)現(xiàn)個(gè)入門的問題。

不要再苦沒有合適的kafka管理平臺(tái)，給你分享10款kafka管理工具

這10款工具如下：

AKHQ

Kowl

Kafdrop

UI for Apache Kafka

Lenses

CMAK

Confluent CC

Conduktor

LogiKM

kafka-console-ui

如果上面這個(gè)地址可以打開，可以直接去看介紹，下文也不再重復(fù)說明。

關(guān)于前8款的對(duì)比，可以看下面這張圖片，圖片也是于上面，我直接copy過來了（可能有好多同學(xué)打不開上面這個(gè)鏈接，就直接看這張圖片了解了下吧）

關(guān)于這8款工具的介紹，人家說的很清晰了，這里就不再重復(fù)說明了，并且這些工具，大部分我也沒用過，也沒資格評(píng)價(jià)太多。

考慮到很多同學(xué)可能打開github太慢，我下面會(huì)把相關(guān)基本信息整理一下，供大家快速了解，方便選型。

概覽

AKHQ (previously known as KafkaHQ)

開發(fā)語言：后端是java為主

Kowl - A Web UI for Apache Kafka

p.s. github上完整的動(dòng)圖這里上傳失敗，就只放一個(gè)靜態(tài)的截圖了，如果可以打開github，建議打開下面的地址直接看吧。

但是這個(gè)并不是所有功能都是免費(fèi)，有部分功能是商業(yè)版才有：

開發(fā)語言：后端是go為主

Kafdrop – Kafka Web UI

開發(fā)語言：后端以java為主

要求jdk11或更高版本

UI for Apache Kafka – Free Web UI for Apache Kafka

開發(fā)語言：后端以java為主

要求jdk13或更高版本

Lenses.io

Apache Kafka 和 Kubernetes 的實(shí)時(shí)應(yīng)用程序和數(shù)據(jù)操作 #DataOps 門戶。

CMAK (Cluster Manager for Apache Kafka, previously known as Kafka Manager)

這個(gè)想必很多同學(xué)都知道，原來的名字就是kafka manager。

開發(fā)語言：后端以scala為主

Confluent Inc.

Apache

Conduktor

一個(gè)商業(yè)版本的桌面客戶端

官網(wǎng)找到一個(gè)這樣的圖片，湊合看吧：

LogiKM

滴滴開源的一站式Apache Kafka集群指標(biāo)監(jiān)控與運(yùn)維管控平臺(tái)。

也是分社區(qū)版和商業(yè)版的。

這個(gè)建議直接看github說明吧，都是中文，內(nèi)容清晰，相關(guān)的資料也都有。

我也簡單的了解了下，有個(gè)邏輯集群的概念，對(duì)于規(guī)模比較大的kafka集群管理還是挺好的，不過，這里比較高端的特性都是不開源的，必須商業(yè)版才能用。

開發(fā)語言：后端以java為主

kafka-console-ui(kafka可視化管理平臺(tái))

一款輕量級(jí)的kafka可視化管理平臺(tái)，安裝配置快捷、簡單易用。界面風(fēng)格有點(diǎn)類似rocketmq-console。

這款權(quán)當(dāng)是“王婆賣瓜，自賣自夸”吧，一個(gè)小工具，如果剛接觸kafka的同學(xué)或者是中小型集群，想找個(gè)簡單易用的，可以考慮一下。

開發(fā)語言：后端以java和scala為主

參考鏈接：

Golang kafka簡述和操作（sarama同步異步和消費(fèi)組）

一、Kafka簡述

1. 為什么需要用到消息隊(duì)列

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

解耦：在耦合太高的場(chǎng)景，多個(gè)任務(wù)要對(duì)同一個(gè)數(shù)據(jù)進(jìn)行操作消費(fèi)的時(shí)候，會(huì)導(dǎo)致一個(gè)任務(wù)的處理因?yàn)榱硪粋€(gè)任務(wù)對(duì)數(shù)據(jù)的操作變得及其復(fù)雜。

緩沖：當(dāng)遇到突發(fā)大流量的時(shí)候，消息隊(duì)列可以先把所有消息有序保存起來，避免直接作用于系統(tǒng)主體，系統(tǒng)主題始終以一個(gè)平穩(wěn)的速率去消費(fèi)這些消息。

2.為什么選擇kafka呢？

這沒有絕對(duì)的好壞，看個(gè)人需求來選擇，我這里就抄了一段他人總結(jié)的的優(yōu)缺點(diǎn)，可見原文

kafka的優(yōu)點(diǎn)：

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

kafka的缺點(diǎn)：

1.由于是批量發(fā)送，所以數(shù)據(jù)達(dá)不到真正的實(shí)時(shí)2.對(duì)于mqtt協(xié)議不支持3.不支持物聯(lián)網(wǎng)傳感數(shù)據(jù)直接接入4.只能支持統(tǒng)一分區(qū)內(nèi)消息有序，無法實(shí)現(xiàn)全局消息有序5.監(jiān)控不完善，需要安裝插件6.需要配合zookeeper進(jìn)行元數(shù)據(jù)管理7.會(huì)丟失數(shù)據(jù)，并且不支持事務(wù)8.可能會(huì)重復(fù)消費(fèi)數(shù)據(jù)，消息會(huì)亂序，可用保證一個(gè)固定的partition內(nèi)部的消息是有序的，但是一個(gè)topic有多個(gè)partition的話，就不能保證有序了，需要zookeeper的支持，topic一般需要人工創(chuàng)建，部署和維護(hù)一般都比mq高

3. Golang 操作kafka

3.1. kafka的環(huán)境

網(wǎng)上有很多搭建kafka環(huán)境教程，這里就不再搭建，就展示一下kafka的環(huán)境，在kubernetes上進(jìn)行的搭建，有需要的私我，可以發(fā)yaml文件

3.2. 第三方庫

github.com/Shopify/sarama // kafka主要的庫*github.com/bsm/sarama-cluster // kafka消費(fèi)組

3.3. 消費(fèi)者

單個(gè)消費(fèi)者

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所有的分區(qū)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){//為每個(gè)分區(qū)開一個(gè)go協(xié)程去取值formsg :=rangepc.Messages() {//阻塞直到有值發(fā)送過來，然后再繼續(xù)等待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()}

消費(fèi)組

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 并不是實(shí)時(shí)寫入kafka，有可能在程序crash時(shí)丟掉未提交的offset}}funcmain(){goconsumerCluster()}

3.4. 生產(chǎn)者

同步生產(chǎn)者

packagemainimport("fmt""github.com/Shopify/sarama")funcmain(){? config := sarama.NewConfig()? config.Producer.RequiredAcks = sarama.WaitForAll//賦值為-1：這意味著producer在follower副本確認(rèn)接收到數(shù)據(jù)后才算一次發(fā)送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//寫到隨機(jī)分區(qū)中，默認(rèn)設(shè)置8個(gè)分區(qū)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("分區(qū)ID:%v, offset:%v \n", pid, offset)}

異步生產(chǎn)者

funcasyncProducer(){? config := sarama.NewConfig()? config.Producer.Return.Successes =true//必須有這個(gè)選項(xiàng)config.Producer.Timeout =5* time.Second? p, err := sarama.NewAsyncProducer(strings.Split("172.20.3.13:30901",","), config)deferp.Close()iferr !=nil{return}//這個(gè)部分一定要寫，不然通道會(huì)被堵塞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. 結(jié)果展示-

同步生產(chǎn)打印：

分區(qū)ID:0,offset:90

消費(fèi)打印：

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

異步生產(chǎn)打印：

async:7272async:7616async:998

消費(fèi)打印：

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

開源數(shù)據(jù)統(tǒng)計(jì)平臺(tái) -- GoAnalytics

本項(xiàng)目用于移動(dòng)端的數(shù)據(jù)統(tǒng)計(jì)，項(xiàng)目地址： 。開源的數(shù)據(jù)統(tǒng)計(jì)countly做的很好，但是基礎(chǔ)免費(fèi)版的功能實(shí)在不夠看，因此我就決定用go語言來寫了這個(gè)項(xiàng)目，一來可以在實(shí)踐中學(xué)習(xí)go語言，二來也可以開發(fā)功能完整的開源平臺(tái)。該項(xiàng)目正在開發(fā)中，歡迎有興趣的gopher一起參與。

數(shù)據(jù)存儲(chǔ)方面使用的是mongodb。由于數(shù)據(jù)統(tǒng)計(jì)業(yè)務(wù)幾乎不涉及到事務(wù)以及嚴(yán)格的一致性場(chǎng)景，而且mongodb的自動(dòng)分片功能可以支撐較大的數(shù)據(jù)量。使用大數(shù)據(jù)的存儲(chǔ)組件的話就太過于重了。因此選用mongodb。

業(yè)務(wù)邏輯整體基于事件的發(fā)布訂閱。當(dāng)收到客戶端請(qǐng)求， frontend 會(huì)對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行處理，然后發(fā)布響應(yīng)的事件。 backend 收到事件后進(jìn)行統(tǒng)計(jì)處理。

后臺(tái)展示基于Vue-Admin-Template開發(fā)，本人前端能力基本就是依葫蘆畫瓢，希望有前端大神來開發(fā)后臺(tái)頁面，項(xiàng)目地址:

目前客戶端API僅有2個(gè)。一個(gè)是上報(bào) openApp 打開APP時(shí)間，一個(gè)是上報(bào) usageTime 一次啟動(dòng)使用時(shí)長事件。SDK方面也需要移動(dòng)端的大神開發(fā)，感興趣的大佬可以一起開發(fā)。

下面放一點(diǎn)后臺(tái)頁面的效果圖：

GoAnalytics是基于go實(shí)現(xiàn)的一個(gè)數(shù)據(jù)統(tǒng)計(jì)平臺(tái)，用于統(tǒng)計(jì)移動(dòng)端的數(shù)據(jù)指標(biāo)，比如啟動(dòng)次數(shù)、用戶增長、活躍用戶、留存等指標(biāo)分析。前端數(shù)據(jù)展示項(xiàng)目是 goanalytics-web 。目前正在積極開發(fā)中，歡迎提交新的需求和pull request。

Go版本需要支持module，本地開發(fā)測(cè)試

cmd/goanalytics_kafka 和 goanalytics_rmq 是分別基于 kafka 和 rocketmq 的發(fā)布訂閱功能做的數(shù)據(jù)發(fā)布

和訂閱處理，橫向擴(kuò)展能力比 local 高。另外由于 rocketmq 還沒有原生基于 go 的客戶端（原生客戶端正在開發(fā)中

2.0.0 road map )，可能會(huì)存在問題。

項(xiàng)目結(jié)構(gòu)

├── README.md

├── api

│ ├── authentication 用戶認(rèn)證、管理API

│ ├── middlewares GIN 中間件

│ └── router API route

├── cmd

│ ├── account 生成admin賬號(hào)命令

│ ├── analytic_local 不依賴消息系統(tǒng)的goanalytics

│ ├── goanalytics_kafka 基于kafak的goanalytics

│ ├── goanalytics_rmq 基于rocketmq的goanalytics

│ └── test_data 生成測(cè)試數(shù)據(jù)命令

├── common

│ └── data.go

├── conf 配置

│ └── conf.go

├── event

│ ├── codec 數(shù)據(jù)編解碼

│ └── pubsub 消息發(fā)布訂閱

├── go.mod

├── go.sum

├── metric 所有的統(tǒng)計(jì)指標(biāo)在這里實(shí)現(xiàn)

│ ├── init.go

│ └── user 用戶相關(guān)指標(biāo)的實(shí)現(xiàn)

├── schedule

│ └── schedule.go 定時(shí)任務(wù)調(diào)度

├── storage 存儲(chǔ)模塊

│ ├── counter.go 計(jì)數(shù)器接口

│ ├── data.go

│ └── mongodb 基于mongodb實(shí)現(xiàn)的存儲(chǔ)及計(jì)數(shù)器

└── utils

├── date.go

├── date_test.go

├── errors.go

└── key.go

filebeat采集日志到kafka配置及使用

Filebeat是elastic公司beats系列工具中的一個(gè)，主要用于收集本地日志。

在服務(wù)器上安裝后，filebeat會(huì)監(jiān)控日志目錄或者指定的日志文件，追蹤讀取這些文件（追蹤文件的變化，不停的讀），并且轉(zhuǎn)發(fā)這些信息到配置文件中指定的輸出端（例如：elasticsearch,logstarsh或kafka）。

Filebeat使用go語言開發(fā)，使用時(shí)沒有其他依賴，比logstash-forworder輕量，不會(huì)占用部署服務(wù)器太多的資源。

filebeat的工作流程：當(dāng)你開啟filebeat程序的時(shí)候，它會(huì)啟動(dòng)一個(gè)或多個(gè)探測(cè)器（prospectors）去檢測(cè)你指定的日志目錄或文件，對(duì)于探測(cè)器找出的每一個(gè)日志文件，filebeat啟動(dòng)收割進(jìn)程（harvester），每一個(gè)收割進(jìn)程讀取一個(gè)日志文件的新內(nèi)容，并發(fā)送這些新的日志數(shù)據(jù)到處理程序（spooler），處理程序會(huì)集合這些事件，最后filebeat會(huì)發(fā)送集合的數(shù)據(jù)到你指定的地點(diǎn)。

2.配置filebeat

配置filebeat需要編輯filebeat的配置文件，不同安裝方式，配置文件的存放路徑有一些不同， 對(duì)于 rpm 和 deb的方式, 配置文件路徑的是 /etc/filebeat/filebeat.yml，對(duì)于壓縮包的方式，配置文件存在在解壓目錄下（例如：我是在home目錄下進(jìn)行的解壓，那么配置文件的路徑就應(yīng)該是~/filebeat-6.2.4-linux-x86_64/filebeat.yml）。

由于我的預(yù)期目標(biāo)是將filebeat收集的日志發(fā)送到kafka,所以配置output就選擇了kafka。讀者可根據(jù)自己的使用場(chǎng)景，配置output。

例子中的配置將對(duì)/var/log目錄下所有以.log結(jié)尾的文件進(jìn)行采集。

3.啟動(dòng)

本文中只是為滿足需求對(duì)filebeat進(jìn)行了最基本的配置。filebeat的很多重要的配置和特性并沒有體現(xiàn)（例如：模塊，多行消息），讀者如果需要更深入的了解請(qǐng)參考： 。

歡迎大家在評(píng)論區(qū)討論使用過程的心得和疑惑。

聊聊golang的zap的ZapKafkaWriter

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

WriteSyncer內(nèi)嵌了io.Writer接口，定義了Sync方法；Sink接口內(nèi)嵌了zapcore.WriteSyncer及io.Closer接口；ZapKafkaWriter實(shí)現(xiàn)Sink接口及zapcore.WriteSyncer接口，其Write方法直接將data通過kafka發(fā)送出去。

名稱欄目：go語言kafka,Go語言之父
路徑分享：http://chinadenli.net/article0/hescoo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供App設(shè)計(jì)、外貿(mào)建站、網(wǎng)站排名、建站公司、品牌網(wǎng)站制作、網(wǎng)站設(shè)計(jì)公司

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)