MQTT概念及其原理

簡(jiǎn)述

MQTT(Message Queuing Telemetry Transport，消息隊(duì)列遙測(cè)傳輸協(xié)議），是一種基于發(fā)布/訂閱（publish/subscribe)模式的"輕量
級(jí)"通訊協(xié)議，該協(xié)議構(gòu)建于TCP/IP協(xié)議上，由IBM在1999年發(fā)布。MQTT最大優(yōu)點(diǎn)在于，可以以極少的代碼和有限的帶寬，為連接遠(yuǎn)程設(shè)備提供實(shí)時(shí)可靠的消息服務(wù)。作為一種低開銷、低帶寬占用的即時(shí)通訊協(xié)議，使其在物聯(lián)網(wǎng)、小型設(shè)備、移動(dòng)應(yīng)用等方面有較廣泛的應(yīng)用。MQTT是一個(gè)基于客戶端-服務(wù)器的消息發(fā)布/訂閱傳輸協(xié)議。

MQTT協(xié)議是輕量、簡(jiǎn)單、開放和易于實(shí)現(xiàn)的，這些特點(diǎn)使它適用范圍非常廣泛。在很多情況下，包括受限的環(huán)境中，如：機(jī)器與機(jī)器（M2M)通信和物聯(lián)網(wǎng)(IoT)。其在，通過衛(wèi)星鏈路通信傳感器、偶爾撥號(hào)的醫(yī)療設(shè)備、智能家居、及一些小型化設(shè)備中已廣泛使用。

MQTT Broker算是一個(gè)服務(wù)器中轉(zhuǎn)站

設(shè)計(jì)規(guī)范

由于物聯(lián)網(wǎng)的環(huán)境是非常特別的，所以MQTT遵循以下設(shè)計(jì)原則：

• 精簡(jiǎn)，不添加可有可無的功能
• 發(fā)布/訂閱（Pub/Sub）模式，方便消息在傳感器之間傳遞
• 允許用戶動(dòng)態(tài)創(chuàng)建主題，零運(yùn)維成本
• 把傳輸量降到最低以提高傳輸效率
• 把低帶寬、高延遲、不穩(wěn)定的網(wǎng)絡(luò)等因素考慮在內(nèi)
• 支持連續(xù)的會(huì)話控制
• 理解客戶端計(jì)算能力可能很低
• 提供服務(wù)質(zhì)量管理
• 假設(shè)數(shù)據(jù)不可知，不強(qiáng)求傳輸數(shù)據(jù)的類型與格式，保持靈活性。

主要特性

MQTT協(xié)議工作在低帶寬、不可靠的網(wǎng)絡(luò)的遠(yuǎn)程傳感器和控制設(shè)備通訊而設(shè)計(jì)的協(xié)議，它具有以下主要的幾項(xiàng)特性：

• 使用發(fā)布/訂閱消息模式，提供一對(duì)多的消息發(fā)布，解除應(yīng)用程序耦合。

這一點(diǎn)很類似于XMPP，但是MQTT的信息冗余遠(yuǎn)小于XMPP，,因?yàn)閄MPP使用XML格式文本來傳遞數(shù)據(jù)。可擴(kuò)展消息處理現(xiàn)場(chǎng)協(xié)議(eXtensible Messaging and Presence Protocol,XMPP)是一種基于可擴(kuò)展標(biāo)記語言(eXtensible Markup Language,XML)的近端串流式即時(shí)通信協(xié)議。

• 對(duì)負(fù)載內(nèi)容屏蔽的消息傳輸。

使用TCP/IP提供網(wǎng)絡(luò)連接。主流的MQTT是基于TCP連接進(jìn)行數(shù)據(jù)推送的，但是同樣有基于UDP的版本，叫做MQTT-SN。這兩種版本由于基于不同的連接方式，優(yōu)缺點(diǎn)自然也就各有不同了。

• 有三種消息發(fā)布服務(wù)質(zhì)量：

“至多一次”，消息發(fā)布完全依賴底層TCP/IP網(wǎng)絡(luò)。會(huì)發(fā)生消息丟失或重復(fù)。這一級(jí)別可用于如下情況，環(huán)境傳感器數(shù)據(jù)，丟失一次讀記錄無所謂，因?yàn)椴痪煤筮€會(huì)有第二次發(fā)送。這一種方式主要普通APP的推送，倘若你的智能設(shè)備在消息推送時(shí)未聯(lián)網(wǎng)，推送過去沒收到，再次聯(lián)網(wǎng)也就收不到了。

“至少一次”，確保消息到達(dá)，但消息重復(fù)可能會(huì)發(fā)生。

“只有一次”，確保消息到達(dá)一次。在一些要求比較嚴(yán)格的計(jì)費(fèi)系統(tǒng)中，可以使用此級(jí)別。在計(jì)費(fèi)系統(tǒng)中，消息重復(fù)或丟失會(huì)導(dǎo)致不正確的結(jié)果。這種最高質(zhì)量的消息發(fā)布服務(wù)還可以用于即時(shí)通訊類的APP的推送，確保用戶收到且只會(huì)收到一次。

• 小型傳輸，開銷很小（固定長(zhǎng)度的頭部是2字節(jié)），協(xié)議交換最小化，以降低網(wǎng)絡(luò)流量。

這就是為什么在介紹里說它非常適合"在物聯(lián)網(wǎng)領(lǐng)域，傳感器與服務(wù)器的通信，信息的收集"，要知道嵌入式設(shè)備的運(yùn)算能力和帶寬都相對(duì)薄弱，使用這種協(xié)議來傳遞消息再適合不過了。

• 使用Last Will和Testament特性通知有關(guān)各方客戶端異常中斷的機(jī)制。

Last Will：即遺言機(jī)制，用于通知同一主題下的其他設(shè)備發(fā)送遺言的設(shè)備已經(jīng)斷開了連接。

Testament：遺囑機(jī)制，功能類似于Last Will。

MQTT協(xié)議實(shí)現(xiàn)方式

實(shí)現(xiàn)MQTT協(xié)議需要客戶端和服務(wù)器端通訊完成，在通訊過程中，MQTT協(xié)議中有三種身份：發(fā)布者（Publish）、代理（Broker) （服務(wù)
器）、訂閱者(Subscribe）。其中，消息的發(fā)布者和訂閱者都是客戶端，消息代理是服務(wù)器，消息發(fā)布者可以同時(shí)是訂閱者。
MQTT傳輸?shù)南⒎譃椋褐黝}（Topic）和負(fù)載（payload）兩部分：

• Topic可以理解為消息的類型，訂閱者訂閱(Subscribe）后，就會(huì)收到該主題的消息內(nèi)容（payload)
• payload，可以理解為消息的內(nèi)容，是指訂閱者具體要使用的內(nèi)容。

MQTT客戶端

一個(gè)使用MQTT協(xié)議的應(yīng)用程序或者設(shè)備，它總是建立到服務(wù)器的網(wǎng)絡(luò)連接。客戶端可以：

• 發(fā)布其他客戶端可能會(huì)訂閱的信息
• 訂閱其它客戶端發(fā)布的消息
• 退訂或刪除應(yīng)用程序的消息
• 斷開與服務(wù)器連接。

MQTT服務(wù)器

MQTT服務(wù)器以稱為"消息代理"(Broker)，可以是一個(gè)應(yīng)用程序或一臺(tái)設(shè)備。它是位于消息發(fā)布者和訂閱者之間，它可以：

• 接受來自客戶的網(wǎng)絡(luò)連接

• 接受客戶發(fā)布的應(yīng)用信息

• 處理來自客戶端的訂閱和退訂請(qǐng)求

• 向訂閱的客戶轉(zhuǎn)發(fā)應(yīng)用程序消息。

MQTT協(xié)議中的方法

MQTT協(xié)議中定義了一些方法（也被稱為動(dòng)作），來于表示對(duì)確定資源所進(jìn)行操作。這個(gè)資源可以代表預(yù)先存在的數(shù)據(jù)或動(dòng)態(tài)生成數(shù)據(jù),
的這取決于服務(wù)器的實(shí)現(xiàn)。通常來說，資源指服務(wù)器上的文件或輸出。主要方法有：

• Connect。等待與服務(wù)器建立連接。
• disconnet. 等的MQT客戶定完流所借的工作，并與服務(wù)都斷開工CP/P會(huì)活。
• Subscribe。等待完成訂閱。
• UnSubscribe。等待服務(wù)器取消客戶端的一個(gè)或多個(gè)topics訂閱。
• Publish。MQTT客戶端發(fā)送消息請(qǐng)求，發(fā)送完成后返回應(yīng)用程序線程。

使用MQTT消息通信-Mosquitto搭建

MQTT協(xié)議廣泛用于物聯(lián)網(wǎng)設(shè)備的消息傳輸，關(guān)于MQTT和MQTT的使用，MQTT的服務(wù)軟件Mosquitto的搭建和使用，MQTT的python開發(fā)軟件包paho-mqtt的使用，以及我們的物聯(lián)網(wǎng)平臺(tái)如何集成MQTT消息服務(wù)及其代碼開發(fā)。

MQTT物聯(lián)網(wǎng)應(yīng)用場(chǎng)景設(shè)計(jì)

• 場(chǎng)景一： 云端訂閱設(shè)備。設(shè)備采用MQTT協(xié)議發(fā)布（publish）數(shù)據(jù)，我們?cè)贫说膇otplus平臺(tái)需要接收數(shù)據(jù)并進(jìn)行保存。我們可以理
  解為，設(shè)備發(fā)布數(shù)據(jù)，云端訂閱保存數(shù)據(jù)。
• **場(chǎng)景二：**向設(shè)備下發(fā)指令數(shù)據(jù)，設(shè)備側(cè)訂閱。用戶一般采用移動(dòng)APP應(yīng)用軟件控制物聯(lián)網(wǎng)設(shè)備，如果移動(dòng)APP應(yīng)用也是采用MQTT協(xié)議發(fā)布（publish）指令數(shù)據(jù)，設(shè)備側(cè)訂閱到數(shù)據(jù)很容易。如果移動(dòng)APP應(yīng)用調(diào)用的是restfulapi接口，我們需要在接口的業(yè)務(wù)邏輯里增加發(fā)布(publish）指令的動(dòng)作。（即在接口設(shè)置發(fā)布mqtt的發(fā)布程序）
• **場(chǎng)景三：**MQTT動(dòng)作。怎么理解MQTT動(dòng)作呢？簡(jiǎn)單理解就是MQTT發(fā)布一個(gè)消息出去。MQTT動(dòng)作的應(yīng)用場(chǎng)景是什么呢？舉例來說，在觸發(fā)的情況下，當(dāng)傳感器A測(cè)量出的溫度大于28攝氏度時(shí)，我們希望觸發(fā)系統(tǒng)自動(dòng)發(fā)布一個(gè)溫度設(shè)置的指令給到空調(diào)，系統(tǒng)發(fā)布的動(dòng)作，我們可以采用MQTT協(xié)議發(fā)布（publish）指令。這個(gè)就是我們理解的MQTT動(dòng)作及其使用場(chǎng)景。

Mosquitto

Eclipse Mosquitto是一個(gè)開源消息代理，實(shí)現(xiàn)了MQTT協(xié)議版本3.1和3.1.1。Mosquitto輕量，適用于低功耗單板計(jì)算機(jī)到完整服務(wù)器的
所有設(shè)備。Mosquitto項(xiàng)目還提供了用于實(shí)現(xiàn)MQTT客戶端的C庫以及非常受歡迎的mosquitt_pub和mosquitto_sub命令行MQTT客戶端。paho-mqtt就是實(shí)現(xiàn)MQTT客戶端的python庫。

unbutu上安裝

apt install mosquitto

初步使用MQTT的python庫paho-mqtt

我們?cè)谏洗握n講到的iotplus使,用MQTT的三個(gè)場(chǎng)景還是非常復(fù)雜的，在正式開發(fā)MQTT的功能之前，我們需要掌握MQTT的python軟件庫的使用。我們已經(jīng)搭建MQTT服務(wù)器，這次課主要講解paho-mqtt的使用。

• 安裝

pip3 install paho-maqtt -i

mqtt_pub.py

# python 3.6

import random
import time

from paho.mqtt import client as mqtt_client


broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'


def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client


def publish(client):
    msg_count = 0
    while True:
        time.sleep(1)
        msg = f"messages: {msg_count}"
        result = client.publish(topic, msg)
        # result: [0, 1]
        status = result[0]
        if status == 0:
            print(f"Send `{msg}` to topic `{topic}`")
        else:
            print(f"Failed to send message to topic {topic}")
        msg_count += 1


def run():
    client = connect_mqtt()
    client.loop_start()
    publish(client)


if __name__ == '__main__':
    run()

mqtt_sub.py文章來源地址http://www.zghlxwxcb.cn/news/detail-691741.html

# python3.6

import random

from paho.mqtt import client as mqtt_client


broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 100)}'


def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client


def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")

    client.subscribe(topic)
    client.on_message = on_message


def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()


if __name__ == '__main__':
    run()

到了這里，關(guān)于物聯(lián)網(wǎng)通信協(xié)議-MQTT及使用python實(shí)現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！