05--MQTT物聯網協議

一、MQTT的概念

MQTT 協議快速入門 2025：基礎知識和實用教程 | EMQ

1.MQTT（Message Queuing Telemetry Transport）是一種輕量級、基于發布-訂閱模式的消息傳輸協議，適用于資源受限的設備和低帶寬、高延遲或不穩定的網絡環境。它在物聯網應用中廣受歡迎，能夠實現傳感器、執行器和其它設備之間的高效通信。

2.訂閱與發布規則

二、MQTT 代理服務器

三、Mosquitto 代理服務器

Eclipse Mosquitto?也是一款開源的 MQTT Broker，兼容 MQTT 協議的 5.0、3.1.1 和 3.1 版本。Mosquitto 體積小巧，既可以運行在低功耗的單板計算機上，也可以部署在企業級服務器上。它采用 C 語言編寫，可以用 C 庫實現 MQTT 客戶端。它支持 Windows、Mac、Linux 和 Raspberry Pi 等多種平臺，為每個平臺提供了方便安裝的二進制文件。最新版本還增加了一個認證和授權插件 “mosquitto-go-auth”，以及一個用于管理 Mosquitto 實例的 Web 用戶界面。此外，它還提供了一個 PHP 包裝器 “Mosquitto-PHP”，可以方便地在 PHP 中開發 MQTT 客戶端。

Eclipse Mosquitto 官方網站

GitHub - eclipse-mosquitto/mosquitto: Eclipse Mosquitto - An open source MQTT broker github開源網站

Mosquitto 服務器安裝

1.下載 openssl 加密庫源碼

Old 1.1.1 Releases | OpenSSL Library

2.下載cjson 源碼

GitHub - DaveGamble/cJSON: Ultralightweight JSON parser in ANSI C

3.下載mosquitto 服務器

Index of /files/source/

4.配置&安裝??

將下載好文件放到一個自己知道的路徑，然后用wsl打開解壓到家目錄。不要解壓到共享文件里。

------- openssl加密庫安裝---------
tar  -xvf   openssl-1.1.1q.tar.gz   -C    ~/   #1.解壓源碼 
cd  ~/openssl-1.1.1q/                          #2.進入源碼目錄  
./config                                       #3.默認配置  make   test  -j12                             #4.編譯測試代碼與庫文件sudo  make  install                           #5.安裝  ------cjson庫安裝-------cp  cJSON-master.zip   ~/    #1.拷貝到家目錄cd  ~/                       #2.進入家目錄sudo apt install unzip       #3.安裝解壓工具 unzip  cJSON-master.zip      #4.解壓json源碼 cd  cJSON-master/            #5.進入json源碼make                         #6.編譯源碼sudo  make install           #7.安裝 -----安裝mosquitto代理服務器-----
sudo  apt-get install   g++                     #1.安裝g++編譯器
tar  -xvf  mosquitto-2.0.9.tar.gz   -C   ~/     #2.解壓到家目錄  
cd  ~/mosquitto-2.0.9/                          #3.進入源碼目錄   
make                                            #4.編譯  
sudo  make  install                             #5.安裝 tip💡如果cjson庫或ssl 庫安裝失敗    
cJSON - for client JSON output support. Disable with make WITH_CJSON=no Auto detected with CMake.make WITH_CJSON=no   #去掉cjson openssl (libssl-dev on Debian based systems) - disable with make WITH_TLS=nomake WITH_TLS=no

測試成功如下

四、mosquitto 代理服務器使用

1、訂閱與發布命令

 #拷貝庫文件到系統庫中 sudo  cp  /usr/local/lib/lib*   /lib#拷貝系統的配置文件 cp  /etc/mosquitto/mosquitto.conf.example   ./  //當前文件為自己寫代碼的文件里

2、修改端口

3、關閉防火墻

#開啟代理服務器  嘗試能不能行mosquitto  -c  ./mosquitto.conf.example#訂閱主題   'test/topic' mosquitto_sub -t 'test/topic' -v#發布消息   mosquitto_pub -t 'test/topic' -m 'hello world'

4、發布端函數接口

將官方大的代碼拷貝到自己寫代碼的文件里邊看官方代碼邊寫自己的代碼。

//1初始化MQTT庫，在調用任何函數之前必須要調用該函數 
int mosquitto_lib_init(void);
返回值:MOSQ_ERR_SUCCESS - on success.//2創建一個新的客戶端
struct mosquitto *mosquitto_new(const char *id, bool clean_session, void *obj);
id:客戶端ID，如何為NULL ,系統自動分配一個ID，且clean_session參數必須為 true 
clean_session:  true   代理服務器在客戶端斷開后會清除數據 false  代理服務器在客戶端斷開后會保留數據 
obj:傳遞給回調函數的參數  
返回值:成功   客戶端對象地址      失敗    NULL //3連接MQTT服務器 
int mosquitto_connect(struct mosquitto *mosq, const char *host, int port, int keepalive);
mosq:客戶端對象  
host:服務器IP地址 
port:服務器端口號 👉1883 
keepalive:保持連續ping 包,設置一段時間后發送一個ping給服務器 
返回值： MOSQ_ERR_SUCCESS - on success.//4啟動網絡線程，不斷處理網絡數據   
int mosquitto_loop_start(struct mosquitto *mosq);
mosq:客戶端對象  
返回值： MOSQ_ERR_SUCCESS - on success.//5?重點,難點 : 發布消息 
int mosquitto_publish(struct mosquitto *mosq, //客戶端對象  int *mid,  //消息ID ,設置為 NULL 即可const char *topic, //👍發布的消息主題int payloadlen,   //發布消息長度   between 0 and 268,435,455.const void *payload, //發布的消息int qos,  //👍消息質量  0 ,1 , 2bool retain); //消息保留標記為  true 保留消息

qos 消息質量

MQTT（Message Queuing Telemetry Transport）協議定義了三種消息質量等級（QoS），以確保在不同場景下消息的可靠傳輸。
以下是三種QoS級別的詳細說明：
1. QoS 0（At Most Once，最多一次）
特點：消息最多被傳遞一次，沒有確認機制，消息可能會丟失或重復。
適用場景：適用于對消息完整性要求不高的場景，如天氣更新、實時數據流、傳感器數據等。在這些場景中，即使消息丟失或重復，也不會造成重大影響。
傳輸過程：發送方發送消息后，不等待接收方的確認，也不存儲消息以進行重傳。
2. QoS 1（At Least Once，至少一次）
特點：消息至少被傳遞一次，但可能會出現重復的消息。
適用場景：適用于對消息丟失敏感，但對重復不敏感的場景，如傳感器數據、開關狀態同步等。在這些場景中，確保消息到達比避免重復更重要。
傳輸過程：發送方發送消息后，會等待接收方的確認報文（PUBACK）。如果未收到確認，發送方會重傳消息。接收方收到消息后，會發送PUBACK確認報文。
3. QoS 2（Exactly Once，僅一次）
特點：消息確保只傳遞一次，沒有重復。這是最高級別的QoS，適用于對消息的完整性和順序性要求非常高的場景。
適用場景：適用于金融交易、關鍵命令等場景，在這些場景中，消息的丟失或重復都是不可接受的。
傳輸過程：發送方和接收方通過四次握手過程（PUBREC、PUBREL、PUBCOMP）來確保消息只被接收一次。發送方發送消息后，等待接收方的PUBREC確認，然后發送PUBREL，最后等待接收方的PUBCOMP確認。
QoS級別的選擇
QoS 0：適用于實時性要求高，但對數據丟失和重復容忍度較高的場景。
QoS 1：適用于需要確保消息至少被接收一次，但允許重復的場景。
QoS 2：適用于需要確保消息僅被接收一次，且對實時性要求不高的場景。
注意事項
網絡條件：在網絡條件較差的情況下，建議選擇較低的QoS級別，以減少消息丟失的風險。
系統資源：QoS級別越高，傳輸過程的復雜程度和系統資源消耗也越大。
綜上所述，根據具體的應用場景和需求選擇合適的QoS級別，可以在確保消息可靠傳輸的同時，優化系統性能和資源利用率。

??發布端代碼例子：

#include <mosquitto.h> //??聲明MQTT庫的接口
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>int main(int argc, char *argv[])
{// 1.初始化MQTT庫mosquitto_lib_init();// 2.創建一個客戶端struct mosquitto *mosq = mosquitto_new(NULL, true, NULL);if (mosq == NULL){perror("創建客戶端失敗\n");return 1;}else{printf("創建客戶成功\n");}// 3.連接服務器，超時設置為60秒int ret = mosquitto_connect(mosq, "127.0.0.1", 1883, 60);if (ret != MOSQ_ERR_SUCCESS){perror("連接服務器失敗\n");return 1;}else{printf("連接服務器成功\n");}// 4.啟動網絡線程mosquitto_loop_start(mosq);while (1){printf("請輸入發布的主題和消息\n");char topic[50] = {0};char payload[268] = {0};scanf("%s %s", topic, payload);int ret = mosquitto_publish(mosq, NULL, topic, strlen(payload), payload, 0, false);if (ret != MOSQ_ERR_SUCCESS){perror("發布消息失敗\n");return 1;}else{printf("發布消息成功\n");}}mosquitto_destroy(mosq); //銷毀對象mosquitto_lib_cleanup(); //清空函數庫 
}

5、訂閱端函數接口

//1.初始化MQTT庫，在調用任何函數之前必須要調用該函數 
int mosquitto_lib_init(void);
返回值:MOSQ_ERR_SUCCESS - on success.//2。創建一個新的客戶端
struct mosquitto *mosquitto_new(const char *id, bool clean_session, void *obj);
id:客戶端ID，如何為NULL ,系統自動分配一個ID，且clean_session參數必須為 true 
clean_session:  true   代理服務器在客戶端斷開后會清除數據 false  代理服務器在客戶端斷開后會保留數據 
obj:傳遞給回調函數的參數  
返回值:成功   客戶端對象地址      失敗    NULL //3.連接MQTT服務器 
int mosquitto_connect(struct mosquitto *mosq, const char *host, int port, int keepalive);
mosq:客戶端對象  
host:服務器IP地址 
port:服務器端口號 👉1883 
keepalive:保持連續ping 包,設置一段時間后發送一個ping給服務器 （超時時間）
返回值： MOSQ_ERR_SUCCESS - on success.//4.訂閱主題  
int mosquitto_subscribe(struct mosquitto *mosq, int *mid, const char *sub, int qos);
mosq:客戶端對象
mid:消息ID,設置為NULL不關心
sub:👍需要訂閱的主題 
qos:消息質量 0,1,2
返回值： MOSQ_ERR_SUCCESS - on success.//5.?重點，難點:設置消息回調函數，當訂閱的主題有消息時，會調用該函數
void mosquitto_message_callback_set(struct mosquitto *mosq, 
void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *));
mosq:客戶端對象
on_message:函數指針，指向一個回調函數如下👇
void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{/* This blindly prints the payload, but the payload can be anything so take care. */printf("%s %d %s\n", msg->topic, msg->qos, (char *)msg->payload);
}
mosq:客戶端對象 
obj:創建客戶端時傳遞的參數 
msg:👍消息💡消息結構體 
struct mosquitto_message{int mid;      //發布消息id char *topic;  //主題void *payload; //消息 int payloadlen; //消息長度  int qos;        //通信質量 bool retain;    //保留消息標記
};6.循環接收消息   int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets);mosq:客戶端對象   timeout:超時檢測 ，設置 -1 永久等待

訂閱端代碼例子:

#include <mosquitto.h> //??聲明MQTT庫的接口
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{/* This blindly prints the payload, but the payload can be anything so take care. */printf("%s %d %s\n", msg->topic, msg->qos, (char *)msg->payload);
}int main(int argc, char *argv[])
{// 1.初始化MQTT庫mosquitto_lib_init();// 2.創建一個客戶端struct mosquitto *mosq = mosquitto_new(NULL, true, NULL);if (mosq == NULL){perror("創建客戶端失敗\n");return 1;}else{printf("創建客戶成功\n");}// 3.連接服務器int ret = mosquitto_connect(mosq, "127.0.0.1", 1883, 60);if (ret != MOSQ_ERR_SUCCESS){perror("連接服務器失敗\n");return 1;}else{printf("連接服務器成功\n");}// 4.訂閱一個主題ret = mosquitto_subscribe(mosq, NULL, "test", 0);if (ret != MOSQ_ERR_SUCCESS){perror("訂閱失敗\n");return 1;}else{printf("訂閱成功\n");}// 5.設置消息回調函數mosquitto_message_callback_set(mosq, on_message); // 設置接收消息回調函數// 6.循環接收消息mosquitto_loop_forever(mosq, -1, 1); // 一直循環接收數據mosquitto_destroy(mosq);mosquitto_lib_cleanup();return 0;
}