一混槐、MQTT 簡(jiǎn)介

MQTT協(xié)議（Message Queuing Telemetry Transport），叫做遙信消息隊(duì)列傳輸。 是工作在TCP/IP協(xié)議族上的一種輕量級(jí)的發(fā)布/訂閱協(xié)議阁吝，需要最小的占用空間和帶寬來(lái)連接 IoT 設(shè)備。與 HTTP 的請(qǐng)求/響應(yīng)范式不同械拍，MQTT 是事件驅(qū)動(dòng)的突勇，可以將消息推送到客戶端。這種類型的架構(gòu)將客戶端彼此解耦坷虑，以實(shí)現(xiàn)高度可擴(kuò)展的解決方案甲馋，而數(shù)據(jù)生產(chǎn)者和數(shù)據(jù)消費(fèi)者之間沒(méi)有依賴關(guān)系。

1.MQTT 的主要優(yōu)點(diǎn)

• 開(kāi)放消息協(xié)議迄损，簡(jiǎn)單易實(shí)現(xiàn)
• 發(fā)布訂閱模式定躏，一對(duì)多消息發(fā)布
• 基于TCP/IP網(wǎng)絡(luò)連接
• 1字節(jié)固定報(bào)頭，2字節(jié)心跳報(bào)文，報(bào)文結(jié)構(gòu)緊湊
• 消息QoS支持痊远，可靠傳輸保證

2.MQTT 的基本概念

MQTT 的核心是MQTT 代理和MQTT 客戶端垮抗。代理負(fù)責(zé)在發(fā)送方和合法接收方之間分發(fā)消息。MQTT 客戶端向代理發(fā)布消息拗引，其他客戶端可以訂閱該代理以接收消息借宵。每個(gè) MQTT 消息都包含一個(gè)主題》鳎客戶端向特定主題發(fā)布消息壤玫，MQTT 客戶端訂閱他們想要接收的主題。MQTT 代理使用主題和訂閱者列表將消息分派到適當(dāng)?shù)目蛻舳恕?/p>

image

MQTT傳輸?shù)南⒎譃椋褐黝}（Topic）和負(fù)載（payload）兩部分：

• Topic哼凯，可以理解為消息的類型欲间，訂閱者訂閱（Subscribe）后，就會(huì)收到該主題的消息內(nèi)容（payload）断部；

• payload猎贴，可以理解為消息的內(nèi)容，是指訂閱者具體要使用的內(nèi)容蝴光，最大支持256MB她渴。

MQTT 代理能夠緩沖無(wú)法分派到未連接的 MQTT 客戶端的消息。這對(duì)于網(wǎng)絡(luò)連接不可靠的情況非常有用蔑祟。為了支持可靠的消息傳遞趁耗，該協(xié)議支持 3 種不同類型的服務(wù)質(zhì)量消息：0 - 最多一次，1 - 至少一次疆虚，2 - 恰好一次苛败。

• Qos0消息發(fā)布訂閱

  image

• Qos1消息發(fā)布訂閱

  image

• Qos2消息發(fā)布訂閱

  image

1）MQTT 客戶端

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

MQTT 客戶端生命周期

• 建立連接：
  • 指定 MQTT Broker 基本信息接入地址與端口
  • 指定傳輸類型是 TCP 還是 MQTT over WebSocket
  • 如果啟用 TLS 需要選擇協(xié)議版本并攜帶相應(yīng)的的證書(shū)
  • Broker 啟用了認(rèn)證鑒權(quán)則客戶端需要攜帶相應(yīng)的 MQTT Username Password 信息
  • 配置客戶端參數(shù)如 keepalive 時(shí)長(zhǎng)罢屈、clean session 回話保留標(biāo)志位、MQTT 協(xié)議版本篇亭、遺囑消息（LWT）等
• 訂閱主題：連接建立成功后可以訂閱主題缠捌，需要指定主題信息
  • 指定主題過(guò)濾器 Topic，訂閱的時(shí)候支持主題通配符 + 與 # 的使用
  • 指定 QoS暗赶，根據(jù)客戶端庫(kù)和 Broker 的實(shí)現(xiàn)可選 Qos 0 1 2
  • 訂閱主題可能因?yàn)榫W(wǎng)絡(luò)問(wèn)題鄙币、Broker 端 ACL 規(guī)則限制而失敗
• 接收消息并處理：
  • 一般是在連接時(shí)指定處理函數(shù)，依據(jù)客戶端庫(kù)與平臺(tái)的網(wǎng)絡(luò)編程模型不同此部分處理方式略有不同
• 發(fā)布消息：向指定主題發(fā)布消息
  • 指定目標(biāo)主題蹂随，注意該主題不能包含通配符 + 或 #十嘿，若主題中包含通配符可能會(huì)導(dǎo)致消息發(fā)布失敗、客戶端斷開(kāi)等情況（視 Broker 與客戶端庫(kù)實(shí)現(xiàn)方式）
  • 指定消息 QoS 級(jí)別
  • 指定消息體內(nèi)容岳锁，消息體內(nèi)容大小不能超出 Broker 設(shè)置最大消息大小
  • 指定消息 Retain 保留消息標(biāo)志位
• 取消訂閱：
  • 指定目標(biāo)主題即可
• 斷開(kāi)連接：
  • 客戶端主動(dòng)斷開(kāi)連接绩衷，服務(wù)器端將發(fā)布遺囑消息（LWT）

Client libraries

C

• Eclipse Paho C
• Eclipse Paho Embedded C
• libmosquitto
• libemqtt - an embedded C client
• MQTT-C - A portable MQTT C client for embedded systems and PCs alike.
• wolfMQTT - Embedded C client
• MQTT over lwIP - MQTT C client for embedded systems using FreeRTOS, lwIP and mbedtls
• libsmartfactory - easy to use library for different Smart Factory/Industry 4.0 technologies including a MQTT client implementation
• libumqtt - A Lightweight and fully asynchronous MQTT client C library based on libev

C++

• Eclipse Paho C++
• libmosquittopp
• Eclipse Paho Embedded C++
• mqtt_cpp - MQTT client and server library based on C++14 and Boost.Asio. It supports MQTT v3.1.1 and v5.

Go

• Eclipse Paho Go
• mqtt by jeffallen
• MQTT

Python

• Eclipse Paho Python - originally the mosquitto Python client
• gmqtt
• nyamuk
• MQTT for twisted python
• HBMQTT
• mqttools

2）MQTT 代理

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

MQTT代理可以：

（1）接受來(lái)自客戶的網(wǎng)絡(luò)連接勿决；

（2）接受客戶發(fā)布的應(yīng)用信息；

（3）處理來(lái)自客戶端的訂閱和退訂請(qǐng)求招盲；

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

如果需要下載MQTT服務(wù)器，可以直接去MQTT Software下載MQTT協(xié)議衍生出來(lái)的各個(gè)不同版本曹货。

常用Servers/ Brokers

• EMQ X

  EMQ X Broker是一個(gè)完全開(kāi)源咆繁、高擴(kuò)展、高可用的分布式 MQTT 消息代理顶籽，適用于物聯(lián)網(wǎng)玩般、M2M 和移動(dòng)應(yīng)用程序，可處理數(shù)千萬(wàn)并發(fā)客戶端礼饱。

  從 3.0 版本開(kāi)始坏为，EMQ X broker 全面支持 MQTT V5.0 協(xié)議規(guī)范，向下兼容 MQTT V3.1 和 V3.1.1镊绪，以及其他通信協(xié)議匀伏，如 MQTT-SN、CoAP蝴韭、LwM2M帘撰、WebSocket 和 STOMP。EMQ X Broker 3.0 版本可以在一個(gè)集群上擴(kuò)展到 10+ 百萬(wàn)并發(fā) MQTT 連接万皿。

• HIVEMQ

  HiveMQ 是一個(gè) MQTT 代理，它從頭開(kāi)始構(gòu)建核行，考慮了最大的可擴(kuò)展性和企業(yè)級(jí)安全性牢硅。它帶有原生 Web 套接字支持和開(kāi)源插件 SDK，以擴(kuò)展其功能或?qū)⑵渑c其他組件集成芝雪。

• Mosquitto

  Mosquitto 是一個(gè)開(kāi)源 MQTT 服務(wù)器减余。還提供公共托管測(cè)試服務(wù)器。

• RabbitMQ

  RabbitMQ 是一個(gè) AMQP 消息代理惩系，帶有一個(gè)MQTT 插件位岔，還提供公共測(cè)試服務(wù)器

3）MQTT協(xié)議中的訂閱、主題堡牡、會(huì)話

• 訂閱（Subscription）

  訂閱包含主題篩選器（Topic Filter）和最大服務(wù)質(zhì)量（QoS）抒抬。訂閱會(huì)與一個(gè)會(huì)話（Session）關(guān)聯(lián)。一個(gè)會(huì)話可以包含多個(gè)訂閱晤柄。每一個(gè)會(huì)話中的每個(gè)訂閱都有一個(gè)不同的主題篩選器擦剑。

• 會(huì)話（Session）

  每個(gè)客戶端與服務(wù)器建立連接后就是一個(gè)會(huì)話，客戶端和服務(wù)器之間有狀態(tài)交互。會(huì)話存在于一個(gè)網(wǎng)絡(luò)之間惠勒，也可能在客戶端和服務(wù)器之間跨越多個(gè)連續(xù)的網(wǎng)絡(luò)連接赚抡。

• 主題名（Topic Name）

  連接到一個(gè)應(yīng)用程序消息的標(biāo)簽，該標(biāo)簽與服務(wù)器的訂閱相匹配纠屋。服務(wù)器會(huì)將消息發(fā)送給訂閱所匹配標(biāo)簽的每個(gè)客戶端涂臣。

• 主題篩選器（Topic Filter）

  一個(gè)對(duì)主題名通配符篩選器，在訂閱表達(dá)式中使用售担，表示訂閱所匹配到的多個(gè)主題赁遗。

• 負(fù)載（Payload）

  消息訂閱者所具體接收的內(nèi)容。

4）連接方式

MQTT客戶端與代理可以基于MQTT-TCP或MQTT-WebSocket協(xié)議建立連接灼舍。

• WebSocket協(xié)議是基于TCP的一種應(yīng)用層網(wǎng)絡(luò)協(xié)議吼和。WebSocket使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡(jiǎn)單，允許服務(wù)端主動(dòng)向客戶端推送數(shù)據(jù)骑素。在WebSocket API中炫乓，瀏覽器和服務(wù)器只需要完成一次握手，兩者之間就可以創(chuàng)建持久性的連接献丑，并進(jìn)行雙向數(shù)據(jù)傳輸末捣。
• WebSocket協(xié)議支持Web瀏覽器（或其他客戶端應(yīng)用程序）與Web服務(wù)器之間的交互，具有較低的開(kāi)銷(xiāo)创橄，便于實(shí)現(xiàn)客戶端與服務(wù)器的實(shí)時(shí)數(shù)據(jù)傳輸箩做。通信通過(guò)TCP端口80或443完成，這在防火墻阻止非Web網(wǎng)絡(luò)連接的環(huán)境下是有益的妥畏。

image

二邦邦、EMQ X Broker使用

1.安裝EMQ X

# 1.Shell 腳本一鍵安裝 (Linux)
wget https://repos.emqx.io/install_emqx.sh 
sudo bash install_emqx.sh

# 2.或者手動(dòng)安裝
# 安裝所需要的依賴包
$ sudo apt update && sudo apt install -y \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg-agent \
    software-properties-common
# 添加 EMQ X 的官方 GPG 密鑰
$ curl -fsSL https://repos.emqx.io/gpg.pub | sudo apt-key add -
# 驗(yàn)證密鑰
$ sudo apt-key fingerprint 3E640D53
pub   rsa2048 2019-04-10 [SC]
    FC84 1BA6 3775 5CA8 487B  1E3C C0B4 0946 3E64 0D53
uid           [ unknown] emqx team <support@emqx.io>
# 添加存儲(chǔ)庫(kù)
$ sudo add-apt-repository \
    "deb [arch=amd64] https://repos.emqx.io/emqx-ce/deb/ubuntu/ \
    ./$(lsb_release -cs) \
    stable"
$ sudo apt update
$ sudo apt install emqx

安裝目錄說(shuō)明

$ tree /var/lib/emqx/
/var/lib/emqx/
├── configs
│   ├── app.2021.09.16.19.06.05.config
│   ├── app.2021.09.16.20.07.27.config
│   ├── vm.2021.09.16.19.06.05.args
│   └── vm.2021.09.16.20.07.27.args
├── emqx_erl_pipes
│   └── emqx@127.0.0.1
│       ├── erlang.pipe.1.r
│       └── erlang.pipe.1.w
├── loaded_modules
├── loaded_plugins
├── mnesia
│   └── emqx@127.0.0.1
│       ├── DECISION_TAB.LOG
│       ├── emqx_activated_alarm.DCD
│       ├── emqx_activated_alarm.DCL
│       ├── emqx_banned.DCD
│       ├── emqx_deactivated_alarm.DCD
│       ├── emqx_mod_delayed.DCD
│       ├── emqx_resource.DCD
│       ├── emqx_rule.DCD
│       ├── emqx_telemetry.DCD
│       ├── emqx_telemetry.DCL
│       ├── LATEST.LOG
│       ├── mqtt_admin.DCD
│       ├── mqtt_admin.DCL
│       ├── mqtt_app.DCD
│       ├── mqtt_app.DCL
│       └── schema.DAT
├── patches
└── scripts

$ tree /usr/lib/emqx/ -L 1
/usr/lib/emqx/
├── bin
├── erts-11.1.8
├── lib
└── releases

$ tree /etc/emqx/ -L 1
/etc/emqx/
├── acl.conf        # EMQ X 默認(rèn) ACL 規(guī)則配置文件
├── certs
├── emqx.conf    # EMQ X 配置文件
├── lwm2m_xml
├── plugins     # EMQ X 擴(kuò)展插件配置文件
├── psk.txt
├── ssl_dist.conf
└── vm.args

2.使用EMQ X Broker

1）啟用EMQ X

# 直接啟動(dòng)
$ sudo emqx start
EMQ X Broker 4.3.5 is started successfully!
# 或
$ sudo systemctl start emqx

# 查看 EMQ X 的狀態(tài)
$ sudo emqx_ctl status
Node 'emqx@127.0.0.1' 4.3.5 is started

# 停止 EMQ X Broker
$ emqx stop
ok

2）EMQ X TCP服務(wù)器

基于/etc/emqx/emqx.conf可以，emqx tcp服務(wù)默認(rèn)監(jiān)聽(tīng)0.0.0.0:1883醉蚁，WebSocket服務(wù)監(jiān)聽(tīng)8083燃辖。參考詳細(xì)配置說(shuō)明。

##--------------------------------------------------------------------
## MQTT/TCP - External TCP Listener for MQTT Protocol

## listener.tcp.$name is the IP address and port that the MQTT/TCP
## listener will bind.
##
## Value: IP:Port | Port
##
## Examples: 1883, 127.0.0.1:1883, ::1:1883
listener.tcp.external = 0.0.0.0:1883

##--------------------------------------------------------------------
## External WebSocket listener for MQTT protocol

## listener.ws.$name is the IP address and port that the MQTT/WebSocket
## listener will bind.
##
## Value: IP:Port | Port
##
## Examples: 8083, 127.0.0.1:8083, ::1:8083
listener.ws.external = 8083

3）Dashboard

訪問(wèn) http://localhost:18083 网棍。默認(rèn)用戶名是 admin黔龟，密碼是 public。

3.EMQ X 客戶端

不同語(yǔ)言可以使用各自的的libraries來(lái)實(shí)現(xiàn)滥玷。

1）MQTT Python 客戶端

sudo pip install paho-mqtt

基于TCP連接emqx broker氏身，發(fā)布emqtt，訂閱testtopic惑畴。

import paho.mqtt.client as mqtt

# 連接成功回調(diào)
def on_connect(client, userdata, flags, rc):
    print('Connected with result code '+str(rc))
    client.subscribe('testtopic/#')

# 消息接收回調(diào)
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

client = mqtt.Client()

# 指定回調(diào)函數(shù)
client.on_connect = on_connect
client.on_message = on_message

# 建立連接
client.connect('broker.emqx.io', 1883, 60)
# 發(fā)布消息
client.publish('emqtt',payload='Hello World',qos=0)

client.loop_forever()

執(zhí)行

python3 emqx-test.py