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

一、Kafka簡述

按需制作可以根據自己的需求進行定制，成都網站設計、做網站構思過程中功能建設理應排到主要部位公司成都網站設計、做網站的運用實際效果公司網站制作網站建立與制做的實際意義

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

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

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

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

2.為什么選擇kafka呢？

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

kafka的優(yōu)點：

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

kafka的缺點：

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

3. Golang 操作kafka

3.1. kafka的環(huán)境

網上有很多搭建kafka環(huán)境教程，這里就不再搭建，就展示一下kafka的環(huán)境，在kubernetes上進行的搭建，有需要的私我，可以發(fā)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所有的分區(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){//為每個分區(qū)開一個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()}

消費組

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副本確認接收到數據后才算一次發(fā)送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//寫到隨機分區(qū)中，默認設置8個分區(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)}

異步生產者

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. 結果展示-

同步生產打印：

分區(qū)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

EMQ功能使用(一) 實現(xiàn)MQTTS協(xié)議

前言

EMQ是帶有SSL功能的，需要進行簡單的配置，才能使用。下面就簡單說一下如何實現(xiàn)自簽證書。

利用OpenSSL簽發(fā)證書

配置到EMQX的emqx.conf

如果啟用單向認證的話，客戶端不需要證書都可以連接。這里的listener.ssl.external.fail_if_no_peer_cert = true 注釋掉就啟用單向認證。啟用雙向認證。那么客戶端就必須導入CA和client的證書才可以連接。

重新啟動EMQX

工具測試

我這里使用EMQ官方出品的MQTTX工具

參考：

【內部分享】MQTT協(xié)議解讀及使用經驗

時間：2018-07-26

Q: 什么是網絡連接？

A: 網絡連接是傳輸層定義的概念，在傳輸層以下只存在網絡數據包的相互交換。

所謂連接，其實也不是在網絡上有一條真實存在的數據通道。只要通信雙方在一段時間內持續(xù)保持數據包交換，就可以視為雙方建立的連接并沒有斷開。

連接的建立是依托于TCP協(xié)議的三次握手，一旦連接已經建立完畢，通信雙方就可以復用這條虛擬通道進行數據交換。如果連接保持長時間工作一直沒有被中斷，那么這樣的TCP連接就俗稱為長連接。

Message Queue Telemetry Transport ，中文直譯： 消息隊列遙測傳輸協(xié)議 。

在MQTT協(xié)議被設計出來的年代，還沒有物聯(lián)網這么時髦的詞匯，當年叫做 遙測設備 。

MQTT協(xié)議真正開始聲名鵲起的原因，是其正好恰恰踩中移動互聯(lián)網發(fā)展的節(jié)拍，為消息推送場景提供了一個既簡便又具有良好擴展性的現(xiàn)成解決方案。

可以看出，MQTT對消息頭的規(guī)定十分精簡， 固定頭部占用空間大小僅為1個字節(jié) ，一個最小的報文占用的空間也 只有兩個字節(jié) （帶一字節(jié)的長度標識位）。

這也是MQTT協(xié)議針對不穩(wěn)定及帶寬低下的網絡環(huán)境做出的特定設計 - - - - 盡可能地節(jié)省一切不必要的網絡開銷 。

Q：為什么MQTT協(xié)議需要心跳報文(PINGREQ, PINGRESP)來維護連接狀態(tài)，只監(jiān)控該TCP的連接狀態(tài)是否可以實現(xiàn)目的？

A: TCP數據傳輸默認的超時時間過長，不符合應用層上細粒度的要求。

TCP數據傳輸超時的情況可分成三種： 服務端斷開 、 客戶端斷開 、 中間網絡斷開 。

在前兩種場景下，若斷開操作是一方主動發(fā)起的，即表示為TCP連接正常結束，雙方走四次揮手流程；若程序異常結束，則會觸發(fā)被動斷開事件，通信另一方也能立刻感知到本次連接所打開的 Socket 出現(xiàn)中斷異常。

唯獨中間網絡的狀態(tài)是通信雙方不能掌握的。 在Linux系統(tǒng)下 ，TCP的連接超時由內核參數來控制，如果通信中的一方沒有得到及時回復，默認會主動再嘗試 6次 。如果還沒有得到及時回應，那么其才會認定本次數據超時。

連帶首次發(fā)包與六次重試，Linux系統(tǒng)下這7次發(fā)包的超時時間分別為 2的0次方 至 2的6次方 ，即1秒、2秒、4秒、8秒、16秒、32秒、64秒，一共127秒。MQTT協(xié)議認為如此長的超時時間對應用層而言粒度太大，因此其在應用層上還單獨設計屬于自身的心跳響應控制。常見的MQTT連接超時多被設定為 60秒 。

擴展知識 - TCP的KeepAlive機制：

由通信中的 報文標識符 ( Packet Identifier )傳達。

Q：僅Publish與Pubrec能保證消息只被投遞一次嗎？

A： 業(yè)務上可以實現(xiàn)，但MQTT協(xié)議并沒有如此設計，原因如下：

每個消息都會擁有屬于自己的報文標識符，但如果需要兩次數據交換就實現(xiàn)消息僅只收到一次，就需要通信雙方記錄下每次使用的報文標識符，并且在處理每一條消息時都需要去重處理，以防消息被重復消費。

但MQTT協(xié)議最初被設計的工作對象是輕量級物聯(lián)設備，為此在協(xié)議的設計中報文標識符被約定為 可重用 ，以減少對設備性能的消耗，換回的代價不得不使用四次網絡數據交換，才能確保消息正好被消費一次。

Q：兩個不同客戶端在發(fā)布與訂閱同一Topic下的消息時，都可以提出通信Qos要求，此時以哪項為基準？

A： 偽命題，故意在分享時埋下坑，等人來踩。

兩個不同客戶端的通信是需要 Broker 進行中轉，而不是直連。因此，通信中存在兩個不同的會話，雙方的Qos要求僅僅作用于它們與 Broker 之間的會話，最終的Qos基準只會向最低要求方看齊。

例：遺囑消息的正確使用方式可參考此篇文章：

雖然可以借助 Retain Message 實現(xiàn)綁定一條消息至某個Topic，以達到消息的暫時保留目的。

但首先 Retain Message 并不是為存儲場景而設計的，再次MQTT協(xié)議并沒有對消息的持久化作出規(guī)定，也就是說Broker重啟后，現(xiàn)有保留消息也將丟失。

Q：兩種特殊消息的使用場景？

A： 遺囑消息，多用于客戶端間獲取互相之間異常斷線的消息通知；

保留消息，可保存 最近一條 廣播通知，多用于公告欄信息的發(fā)布。

Eclipse Mosquitto ：MQTT協(xié)議的最小集實現(xiàn)

有 EMQ , HiveMQ , RabbitMQ MQTT Adapter 等。

Qos=2 消息保障的網絡I/O次數過多，如果不是必需，盡少在程序里使用此類消息。

畢竟當初其設計的目的是為了減少設備的性能占用，但若應用場景并不是物聯(lián)網，而是用于手機、電腦或瀏覽器端等現(xiàn)在已不缺性能的設備上，最好在報文體中，使用UUID生成全局唯一的消息ID，然后自行在業(yè)務解析中判斷此報文是否被消費過。

或者，業(yè)務方在處理消息時保證其被消費的冪等性，也可消除重復消息對系統(tǒng)帶來的影響。

正如MQTT協(xié)議并沒有依賴TCP連接狀態(tài)，自己在應用層協(xié)議上實現(xiàn)心跳報文來控制連接狀態(tài)，業(yè)務方作為MQTT協(xié)議的使用者，也不要完全依賴協(xié)議的工作狀態(tài)，而是依托MQTT協(xié)議建立屬于業(yè)務本身的信息匯報機制，以加強系統(tǒng)的穩(wěn)健性。

Retain Message 可視為客戶端主動拉取的行為。如果業(yè)務系統(tǒng)采用 HTTP+MQTT 雙協(xié)議描述業(yè)務過程，主動拉取的操作也可使用 HTTP 請求替代。

作為一個長連接型的應用，上線前需要根據業(yè)務量級，評估對操作系統(tǒng) 端口數 與 文件描述符 的占用要求，以防服務器資源被打滿。

在服務端的配置文件和客戶端的連接參數中，都擁有 max_inflight_messages 此項配置，來維護 Qos=1 or 2 消息是否被成功消費的狀態(tài)。

MQTT 最初被設計為物聯(lián)網級的通信協(xié)議，因此此參數的默認配額較小（大多數情況下被限制到10至20）。

但如果將MQTT協(xié)議應用至手機、PC或Web端的推送場景時，硬件性能已不在是瓶頸，在實際使用中推薦把此參數調大。

Mosquitto提供Bridge功能，需要我們自己配置。

Bridge 意為橋接，當我們把兩臺Broker橋接在一起時，只需要修改一臺Broker的配置，填上另一臺Broker的運行地址。前一臺Broker將作為客戶端發(fā)布與訂閱后一臺Broker的所有Topic，實現(xiàn)消息互通的目的。

橋接帶來的問題有以下幾點：

我的建議：

Websockets協(xié)議被設計的目的是為瀏覽器提供一個全雙工的通信協(xié)議，方便實現(xiàn)消息推送功能。

在Websockets協(xié)議被設計出來前，受限于HTTP協(xié)議的一問一答模型，消息的推送只能靠輪詢來實現(xiàn)，在資源消耗與時效性保障上，均難以達到令人滿意的效果。

Websockets協(xié)議復用了HTTP協(xié)議的頭部信息，告知瀏覽器接下來的操作將觸發(fā)協(xié)議升級，然后通信雙方繼續(xù)復用HTTP的Header，但報文內容已轉變?yōu)殡p方均接受的新協(xié)議的格式。

Websockets協(xié)議改進了網頁瀏覽中的消息推送的方式，因此被廣泛應用在聊天、支付通知等實時性要求比較高的場合下。

MQTT協(xié)議重點在于 消息隊列的實現(xiàn)，其對消息投遞的方式作出約定，并提供一些額外的通信保障 。

MQTT可采取原生的TCP實現(xiàn)，也有基于Websockets的實現(xiàn)版本。當然后者在網絡字節(jié)的利用率上，不如前者那么精簡。但瀏覽器端無法直接使用TCP協(xié)議，所以就只能基于Websockets協(xié)議開發(fā)。

不過基于Websockets的應用也有方便之處：一是證書不需要額外配置，直接與網站共用一套基礎設施；二是可使用 Nginx 等工具管理流量，與普通HTTP流量可共用一套配置方法。

MQTT非常適合入門，原因如下：

實際的應用場景遠比理想中的復雜，無法一招走遍天下，必須做好取舍。

MQTT協(xié)議在這方面做得很優(yōu)秀，以后工作中可以作為參考，設計好自己負責的業(yè)務系統(tǒng)。

ios 利用mqttclient庫寫客戶端怎么添加ssl證書

iOS7以下版本的國際化方法 創(chuàng)建一個Localizable.strings文件 code中用字符串的地方使用NSLocalizedString，這與系統(tǒng)語言相關 時間顯示，數字，金融與地區(qū)相關，所以需要各類NSFormater, 如NSDateFormatter, NSNumberFormatter … 用命令將所有NSLocalizedString返回的字符串格式化到Localizable.strings里。命令行進入工程目錄（我的工程名是LocalizationTest），運行下面命令： find ./ -name "*.m" -print0 xargs -0 genstrings -o LocalizationTest/en.lproj 其中LocalizationTest/en.lproj根據自己的目錄修改簽SSL證書問題，你可要到沃通CA申請一張免費SSL證書來使用。

mqtt使用WebSocket over TLS(wss)握手失敗

由于網頁運行在https上，所以連接mqtt只能用wss，但是使用自簽證書一直顯示 1015 TLS_HANDSHAKE ，可以判斷為認證階段不通過。在MQTT.fx上面則提示證書非法。后面找了很多資料，終于在一個回到里面找到答案，就記錄下來。

MQTT客戶端軟件MQTT.fx的使用詳解

原地址：

原文鏈接：

MQTT客戶端軟件MQTT.fx的使用詳解

使用說明

mqtt.fx打開后的主頁面如下：

點擊齒輪進行連接設置

本地連接設置：

用戶信息設置：

SSL安全證書設置：

網絡代理設置：

遺囑設置：

連接測試

1、啟動mosquitto

地址，下一步配置使用

2、在主機中打開MQTT.FX軟件

設置連接信息

IP為mosquitto所在的IP，端口號默認為1883。

點擊進行連接

連接成功以后可以進行發(fā)布訂閱。

新聞名稱：go語言mqtt帶證書 go語言 mqtt
當前路徑：http://www.chinadenli.net/article48/doogphp.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供自適應網站、品牌網站制作、網站維護、域名注冊、網站設計、微信小程序

廣告

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