大話 IOT 技術(3) -- MQTT篇

文章目錄

- 前言
- 前情提要
- MQTT介紹
- 組成
- 萬惡的app
- mqtt服務端
- 偽代碼實現
- 開源的力量
- 后話

當你迷茫的時候，請點擊物聯網目錄大綱快速查看前面的技術文章，相信你總能找到前行的方向

前言

本篇將開始講述IOT技術的一個重點，mqtt協議。

我發現有一個物聯網的準則，想分享給大家，名字我都想好了，就叫張氏定理：

手機app端用http協議與服務端（eg.bypass）進行通信
設備端用mqtt協議與mqtt服務端進行通信
服務端（eg.bypass）還要能用mqtt協議與mqtt服務端進行通信

這樣，手機到設備的通信鏈路就算是通了起來。

http協議大家早就司空見慣了，它就像是大貨車，能裝的東西種類很多，本身體積也是比較龐大的，適用于手機這種要實現豐富的功能的大款。

mqtt協議是本文的重點講述的對象，它就像小車，輕便簡捷，我們通常用json作為消息體進行傳遞mqtt信息。

下面我們就進入正文吧。

前情提要

上篇(大話 IOT 技術(2) – 配網篇)我們講了配網的流程，不過與本文關系不大，所以，我們還是先回顧一下經典 IOT 整體架構圖

MQTT介紹

本文我還是沿用我們大話系列的精神，用生動的故事形式，來介紹mqtt技術，所以朋友們不用擔心太枯燥和無聊。

先引用官方的一段介紹作為開場白，詳見 https://mqtt.org/mqtt-specification/：

MQTT is an OASIS standard. The specification is managed by the OASIS MQTT Technical Committee.

MQTT（Message Queuing Telemetry Transport）是一種輕量級的消息傳輸協議，它被設計為易于實現，支持所有消息方向的通信，并為遠程通信環境提供必要的網絡優化。

前面講到過，mqtt協議和http協議很像，它們都是物聯網中的基石。

一般所說xx協議直觀理解都是以xx開頭，比如，http協議以http/https 開頭，例如，url 為 http://test.domain.com/aaa

同理，mqtt協議也是以 mqtt/mqtts開頭，例如，mqtt://test.domain.com/bbb，協議具體怎么實現不是本文關注的（偏底層），有興趣的朋友可以自行百度。

我們主要關注mqtt 應用層，主要里面的幾個關鍵術語，mqtt客戶端，mqtt服務端（又叫broker），連接，mqtt消息，主題（topic）,發布，訂閱。

因為這些和我們工作息息相關，而且已經有開源庫幫我們封裝好了，我們只需要會調用就能實現 IOT 的所有功能。

組成

和眾多C/S架構一樣，由一個 mqtt服務端（server，又叫broker），還有一個或多個 mqtt客戶端（client）組成。

mqtt服務端就像是媽媽，每個客戶端就像是她的孩子，孩子們有心事不會跟其他兄弟姐妹們直接說，都會向媽媽傾訴，由媽媽代為轉述。

媽媽是他們的共同點，在第一篇文章中，我們也去找了手機與設備之間的共同點，只要有共同點，那就有通信的可能，這里也是一樣的。

媽媽只是一個中間商，目的還是要實現各客戶端之間的通信。

比如：

bypass服務 發送 mqtt消息給 mqtt服務端，消息是想查詢設備1的狀態（bypass —> mqtt 服務端）
設備1 從 mqtt服務端獲取消息(訂閱消息)，將自己的狀態參數也寫成一個 mqtt消息，發給 mqtt服務端（mqtt 服務端--->設備1， 設備1 ---> mqtt 服務端）
bypass服務 從 mqtt服務端獲取到了設備1的消息(訂閱消息)，再進行業務處理，更新在手機端顯示（mqtt 服務端 ---> bypass）

這樣，就完成了 bypass服務與設備1 之間的通信。

同理，bypass服務與設備2 之間的通信也是類似。

那么問題又來了，你怎么保證 bypass服務發送的 mqtt消息一定能被設備1 收到呢？設備2 會不會偷看消息？

真是個 good question，這也就是 mqtt服務端 的職責所在。

萬惡的app

首先，我們來講個題外話。

相信大家手機中肯定裝了很多種app，很多萬惡的app在你安裝后就會流氓地要你關注訂閱一些感興趣的頻道，比如，科技，財經，軍事，娛樂…

而它的萬惡是我都感興趣，卻不讓我勾選了，最多只能選幾個，這真的讓人火大，那我一個也不選吧，還不行，你看這整的我只想說一句粗口：……

好吧，我忍了！

于是我手機上多了一個萬惡的app，大概長這樣

有沒有同款的朋友們，請扣1哈

每個關注訂閱的頻道，都會有消息的通知，就像上面那樣，用消息數量顯示，我知道體育頻道沒有消息，我就不會點進去看，我最喜歡的娛樂頻道消息滿天飛，我就會點擊進去瀏覽各種八卦……

雖然萬惡，但其思想與本文要講的 mqtt服務很像，所以我才忍你很久了，曾子曰：“我的忍耐是有限度的……"

上面不經意還是引入了一個mqtt專業術語：訂閱（subscribe），我們再加一個，主題（topic）。

訂閱想必大家很好理解，畢竟爛大街的訂閱加關注，一鍵三連……

主題就是上面的興趣頻道

mqtt 中的訂閱主題，就等同于所說的關注訂閱某個頻道，然后巴巴地等著新消息查看。比如汪蘇瀧的演唱會要來深圳了，什么時候開始搶票，他好帥啊……

這個萬惡的app其實就是充當了服務端的身份，管理這些主題，消息動態，而我就是那個可憐巴巴沒有搶到票的弱小客戶端，悲傷不禁逆流成河……

我們豐富一下流程吧

服務端是那個萬惡的app后端服務，客戶端有某科技博主和我。

發布消息過程：某科技博主寫了一篇文章，標注了科技的標簽，提交后，服務端后臺會處理，會在科技頻道新增一篇科技的文章，并返回成功的響應。

訂閱消息過程：我點擊了娛樂消息想要查看，服務端接收到了我的請求，從數據庫中查詢出標簽為娛樂的消息并返回。

mqtt服務端

類似上面的萬惡app，mqtt服務端也是類似，我們來轉變一下上圖：

服務端變成了mqtt 服務端，客戶端有bypass服務和設備。

發布消息過程：bypass 發布了一個主題為topic1 的消息，mqtt服務端處理發布消息，在topic1新增一條消息，并返回成功的響應。

訂閱消息過程：設備訂閱主題為topic4 的消息，mqtt服務端處理訂閱請求，從主題topic4中獲取其消息并發回給設備，并且客戶端的 on_message方法處理消息。

一般地，主題topic可以設置為含有設備的cid的字符串，更簡單可以用cid 作為主題，既能保證主題唯一，又能方便設備訂閱，這樣每個設備都只需要訂閱與自己 cid 相關的主題。

那么，bypass 可以發送消息到 cid的主題，設備訂閱自己cid 的主題，得到消息后設備再自行處理。

偽代碼實現

上面我們主要關注發布和訂閱的處理，因為這也是mqtt設計的核心，其他的我們可以暫時忽略，如果讓我們自己來設計實現，那我們會怎么做呢？

我認為主要是服務端和客戶端設計實現，首先對它們進行梳理一下

mqtt 服務端：

對客戶端連接的驗證（do_connect），主要校驗賬號密碼或者證書公私鑰是否正確
管理 topic 對應的消息，用隊列保存/讀取，先入先出
處理客戶端的發布請求（do_publish）
處理客戶端的訂閱請求（do_subscribe）

mqtt 客戶端：

連接服務端（connect），連接結果處理（on_connect）
發布消息（publish）
訂閱消息（subscribe），對消息處理（on_message）

在客戶端有 on_connect，on_message 是擴展方法，是留給用戶去實現自定義業務邏輯。

那我們很自然地可以定義 mqtt 的兩個類

下面是我自己實現的mqtt完整的代碼，僅僅代表思路的驗證，不可用于實際生產哦。

import queueclass MqttBroker:def __init__(self) -> None:self.topic_queue_map = {}@staticmethoddef do_connect(**args):print("broker: do_connect 處理客戶端連接請求")# 做一些賬密或證書公私鑰的驗證，這里簡化一下，直接返回truereturn Truedef do_publish(self, msg: dict):print("broker: do_publish 處理客戶端發布請求")topic=msg.get('topic','')q:queue.Queue=self.topic_queue_map.get(topic,queue.Queue())q.put(msg)self.topic_queue_map[topic]=qreturn Truedef do_subscribe(self, topic: str):print("broker: do_subscribe 處理客戶端訂閱請求")q:queue.Queue=self.topic_queue_map.get(topic,queue.Queue())return qclass MqttClient:def __init__(self) -> None:self.conn: MqttBroker = Nonedef connect(self, user_name: str, password: str):print("client: 發起 mqtt connect 請求...")res=MqttBroker.do_connect(user_name=user_name, password=password)if res:self.conn = MqttBroker()self.on_connect(res=res)def publish(self, msg: dict):print(f"client: publish 發布消息{msg}")res=self.conn.do_publish(msg=msg)print("client: publish success !") if res else print("client: publish failed")def subscribe(self, topic):print(f"client: subscribe 訂閱 topic為{topic}的消息")q:queue.Queue=self.conn.do_subscribe(topic=topic)while not q.empty():self.on_message(q.get())def on_connect(self, res):r= '成功' if res else '失敗'print(f"client: on_connect ,連接{r}!")def on_message(self, msg):print(f"client: on_message ,自定義處理消息: {msg}")if __name__ == '__main__':client = MqttClient()# 模擬連接print("模擬連接:")client.connect('user', 'password')# 模擬發布print("\n\n\n模擬發布:")for i in range(3):msg = {'topic': '1001', 'msg': {'playload': f'msg{i}: hello mqtt! '}}client.publish(msg=msg)# 模擬訂閱print("\n\n\n模擬訂閱:")client.subscribe(topic="1001")

執行結果：

模擬連接:
client: 發起 mqtt connect 請求...
broker: do_connect 處理客戶端連接請求
client: on_connect ,連接成功!模擬發布:
client: publish 發布消息{'topic': '1001', 'msg': {'playload': 'msg0: hello mqtt! '}}
broker: do_publish 處理客戶端發布請求
client: publish success !
client: publish 發布消息{'topic': '1001', 'msg': {'playload': 'msg1: hello mqtt! '}}
broker: do_publish 處理客戶端發布請求
client: publish success !
client: publish 發布消息{'topic': '1001', 'msg': {'playload': 'msg2: hello mqtt! '}}
broker: do_publish 處理客戶端發布請求
client: publish success !模擬訂閱:
client: subscribe 訂閱 topic為1001的消息
broker: do_subscribe 處理客戶端訂閱請求
client: on_message ,自定義處理消息: {'topic': '1001', 'msg': {'playload': 'msg0: hello mqtt! '}}
client: on_message ,自定義處理消息: {'topic': '1001', 'msg': {'playload': 'msg1: hello mqtt! '}}
client: on_message ,自定義處理消息: {'topic': '1001', 'msg': {'playload': 'msg2: hello mqtt! '}}