淺析ESP32運行MQTT客戶端進行主題的發布和訂閱的方法
ESP32 MQTT的庫有很多，凌順實驗室（lingshunlab.com）這次主要使用AsyncMQTT_ESP32，以后有機會再更多的MQTT其他庫的使用方法。
前提條件
樹莓派部署本地的MQTT服務端，具體安裝請查看以下連接：
https://lingshunlab.com/book/raspberry-pi/raspberry-pi-install-mosquitto-mqtt-server-and-test-mqtt
ESP32和樹莓派在同一WIFI網絡里面
效果實現
凌順實驗室（lingshunlab.com）在本示例展示了使用兩個ESP32，分別實現發布MQTT的主題消息和訂閱并輸出MQTT的主題內容。當然，可能會問能不能一個ESP32同時又是發布者，又是訂閱者？答案是可以的，因為作為客戶端，都是對中間商做信息交換。
BOM
需要準備2個ESP32，
一個ESP32用于發布，
一個ESP32用于訂閱。
庫的安裝
可以在Arduino IDE的庫管理里搜索并安裝：
點擊菜單欄的「工具」---> 「庫管理」，然后在搜索框中輸入“AsyncMQTT_ESP32”，點擊安裝即可
下圖在我本地已經安裝好了：
又或者在Github中下載，并安裝到Arduino的 "libraries"文件夾里
Github 地址：
https://github.com/khoih-prog/AsyncMQTT_ESP32
程序提點
1， 首先，需要加載AsyncMQTT_ESP32的庫
#include
2，配置MQTT的服務器信息，可以是IP或者域名的方式
//#define MQTT_HOST IPAddress(192, 168, 100, 100) #define MQTT_HOST "broker.emqx.io" // Broker address #define MQTT_PORT 1883
3，設置主題，發布需要主題，訂閱也需要主題
const char *Topic = "lingshunlab/ESP32"; // 主題
4，創建MQTT客戶端的實例
// 創建MQTT客戶端的實例，名為mqttClient AsyncMqttClient mqttClient;
5，認識mqttClient的可用的回調函數
當MQTT觸發特定事件的時候，可以配置自定義的函數
mqttClient.onConnect(onMqttConnect); // 設置 當MQTT連接時的回調函數 mqttClient.onDisconnect(onMqttDisconnect); // 設置 當MQTT斷開連接時的回調函數 mqttClient.onSubscribe(onMqttSubscribe); // 設置 當MQTT訂閱主題時的回調函數 mqttClient.onUnsubscribe(onMqttUnsubscribe); // 設置 當MQTT取消訂閱主題時的回調函數 mqttClient.onMessage(onMqttMessage); // 設置 當MQTT訂閱主題時的回調函數 mqttClient.onPublish(onMqttPublish); // 設置 當取消MQTT訂閱主題時的回調函數 mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設置 MQTT服務器信息
6， 連接MQTT服務器
mqttClient.setServer(MQTT_HOST, MQTT_PORT); //連接MQTT服務器
7，發布主題
通過以下代碼，可以對配置好的主題發布消息
// 發布主題消息 uint16_t packetIdPub = mqttClient.publish(PubTopic, 2, true, "welcome to Lingshunlab.com"); Serial.print("Publisshing at QoS 2, packetId: "); Serial.println(packetIdPub); delay(2000);
8，訂閱主題
通過以下代碼，可以訂閱配置好的主題
// 訂閱MQTT主題，并QoS設置為2
uint16_t packetIdSub = mqttClient.subscribe(SubTopic, 2);
Serial.print("Subscribing at QoS 2, packetId: ");
Serial.println(packetIdSub);
9，當發生主題消息變化的時候的回調函數
mqttClient的回調函數有很多種，請仔細學習查看例子中其他的回調函數。在這里，特別說明一下onMessage的回調函數onMqttMessage（這個函數名稱你可以自己定義，隨喜），里面有不少參數，例如topic，payload等，其中payload即是消息的內容，可以通過輸出顯示。
void onMqttMessage(char* topic, char* payload, const AsyncMqttClientMessageProperties& properties,
const size_t& len, const size_t& index, const size_t& total)
{
(void) payload;
Serial.println("=====On MQTT Message=====");
Serial.println("Publish received.");
Serial.print(" topic: ");
Serial.println(topic);
Serial.print(" qos: ");
Serial.println(properties.qos);
Serial.print(" dup: ");
Serial.println(properties.dup);
Serial.print(" retain: ");
Serial.println(properties.retain);
Serial.print(" len: ");
Serial.println(len);
Serial.print(" index: ");
Serial.println(index);
Serial.print(" total: ");
Serial.println(total);
Serial.print("payload: ");
Serial.println(payload); // 輸出消息內容
}
10，請查看AsyncMQTT_ESP32的官方例子
可以學習到FreeRTOS的多線程如何應用。
發布主題de完整代碼
#include// 配置 WIFI #define WIFI_SSID "***your wifi***" #define WIFI_PASSWORD "***your wifi password***" // 加載AsyncMQTT_ESP32庫 #include // 配置MQTT服務器地址和端口 #define MQTT_HOST IPAddress(192,168,100,100) // Broker IP // #define MQTT_HOST "broker.emqx.io" // Broker address #define MQTT_PORT 1883 const char *PubTopic = "lingshunlab/ESP32"; // 發布消息的主題 AsyncMqttClient mqttClient; // 創建 MQTT客戶端實例 void onMqttConnect(bool sessionPresent) // 編寫對應的回調函數 { Serial.println("=====On MQTT Connect====="); Serial.print("Connected to MQTT broker: "); Serial.print(MQTT_HOST); Serial.print(", port: "); Serial.println(MQTT_PORT); Serial.print("PubTopic: "); Serial.println(PubTopic); } void onMqttDisconnect(AsyncMqttClientDisconnectReason reason) { (void) reason; Serial.println("Disconnected from MQTT."); } void onMqttSubscribe(const uint16_t& packetId, const uint8_t& qos) { Serial.println("Subscribe acknowledged."); Serial.print(" packetId: "); Serial.println(packetId); Serial.print(" qos: "); Serial.println(qos); } void onMqttUnsubscribe(const uint16_t& packetId) { Serial.println("Unsubscribe acknowledged."); Serial.print(" packetId: "); Serial.println(packetId); } void onMqttMessage(char* topic, char* payload, const AsyncMqttClientMessageProperties& properties, const size_t& len, const size_t& index, const size_t& total) { (void) payload; Serial.println("=====On MQTT Message====="); Serial.println("Publish received."); Serial.print(" topic: "); Serial.println(topic); Serial.print(" qos: "); Serial.println(properties.qos); Serial.print(" dup: "); Serial.println(properties.dup); Serial.print(" retain: "); Serial.println(properties.retain); Serial.print(" len: "); Serial.println(len); Serial.print(" index: "); Serial.println(index); Serial.print(" total: "); Serial.println(total); Serial.print("payload: "); Serial.println(payload); // 輸出消息內容 } void onMqttPublish(const uint16_t& packetId) { Serial.println("Publish acknowledged."); Serial.print(" packetId: "); Serial.println(packetId); } void setup() { Serial.begin(115200); while (!Serial && millis() < 5000); delay(500); // 連接WIFI WiFi.begin(WIFI_SSID, WIFI_PASSWORD); while (WiFi.waitForConnectResult() != WL_CONNECTED) { Serial.println("Connection Failed! Rebooting..."); delay(5000); ESP.restart(); // 重啟esp32 } delay(500); mqttClient.onConnect(onMqttConnect); // 設置 當MQTT連接時的回調函數 mqttClient.onDisconnect(onMqttDisconnect); // 設置 當MQTT斷開連接時的回調函數 mqttClient.onMessage(onMqttMessage); // 設置 當MQTT訂閱主題時的回調函數 mqttClient.onPublish(onMqttPublish); // 設置 當取消MQTT訂閱主題時的回調函數 mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設置 MQTT服務器信息 mqttClient.connect(); // 連接 MQTT delay(500); } void loop() { // 發布主題消息 uint16_t packetIdPub = mqttClient.publish(PubTopic, 2, true, "welcome to Lingshunlab.com"); Serial.print("Publisshing at QoS 2, packetId: "); Serial.println(packetIdPub); delay(2000); }
上傳代碼后，程序將會先連接WIFI，然后連接MQTT服務器，再之后每隔2秒發布一個對應主題的消息
訂閱主題de完整代碼
#include// 配置 WIFI #define WIFI_SSID "***your wifi***" #define WIFI_PASSWORD "***your wifi password***" // 加載 AsyncMQTT_ESP32 庫 #include // 配置MQTT服務器地址和端口 #define MQTT_HOST IPAddress(192,168,1,55) // Broker IP // #define MQTT_HOST "broker.emqx.io" // Broker address #define MQTT_PORT 1883 const char *SubTopic = "lingshunlab/ESP32"; // 訂閱的主題 AsyncMqttClient mqttClient; void onMqttConnect(bool sessionPresent) { Serial.println("=====On MQTT Connect====="); Serial.print("Connected to MQTT broker: "); Serial.print(MQTT_HOST); Serial.print(", port: "); Serial.println(MQTT_PORT); Serial.print("PubTopic: "); Serial.println(SubTopic); // 訂閱MQTT主題，并QoS設置為2 uint16_t packetIdSub = mqttClient.subscribe(SubTopic, 2); Serial.print("Subscribing at QoS 2, packetId: "); Serial.println(packetIdSub); } void onMqttDisconnect(AsyncMqttClientDisconnectReason reason) { (void) reason; Serial.println("Disconnected from MQTT."); } void onMqttSubscribe(const uint16_t& packetId, const uint8_t& qos) { Serial.println("=====On MQTT Subscribe====="); Serial.println("Subscribe acknowledged."); Serial.print(" packetId: "); Serial.println(packetId); Serial.print(" qos: "); Serial.println(qos); } void onMqttUnsubscribe(const uint16_t& packetId) { Serial.println("Unsubscribe acknowledged."); Serial.print(" packetId: "); Serial.println(packetId); } void onMqttMessage(char* topic, char* payload, const AsyncMqttClientMessageProperties& properties, const size_t& len, const size_t& index, const size_t& total) { (void) payload; Serial.println("=====On MQTT Message====="); Serial.println("Publish received."); Serial.print(" topic: "); Serial.println(topic); Serial.print(" qos: "); Serial.println(properties.qos); Serial.print(" dup: "); Serial.println(properties.dup); Serial.print(" retain: "); Serial.println(properties.retain); Serial.print(" len: "); Serial.println(len); Serial.print(" index: "); Serial.println(index); Serial.print(" total: "); Serial.println(total); Serial.print("payload: "); Serial.println(payload); } void setup() { Serial.begin(115200); while (!Serial && millis() < 5000); delay(500); // 連接WIFI WiFi.begin(WIFI_SSID, WIFI_PASSWORD); while (WiFi.waitForConnectResult() != WL_CONNECTED) { Serial.println("Connection Failed! Rebooting..."); delay(5000); ESP.restart(); } delay(500); mqttClient.onConnect(onMqttConnect); // 設置 當MQTT連接時的回調函數 mqttClient.onDisconnect(onMqttDisconnect); // 設置 當MQTT斷開連接時的回調函數 mqttClient.onSubscribe(onMqttSubscribe); // 設置 當MQTT訂閱主題時的回調函數 mqttClient.onUnsubscribe(onMqttUnsubscribe); // 設置 當取消MQTT訂閱主題時的回調函數 mqttClient.onMessage(onMqttMessage); // 設置 當MQTT收到主題消息時的回調函數 mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設置 MQTT服務器信息 mqttClient.connect(); // 連接 MQTT delay(500); } void loop() { }
上傳代碼后，程序將會先連接WIFI，然后連接MQTT服務器，當連接MQTT時，則會訂閱主題，之后每隔2秒就會收到主題發布的消息
審核編輯：劉清
聲明：本文內容及配圖由入駐作者撰寫或者入駐合作網站授權轉載。文章觀點僅代表作者本人，不代表電子發燒友網立場。文章及其配圖僅供工程師學習之用，如有內容侵權或者其他違規問題，請聯系本站處理。
舉報投訴
-
回調函數
+關注
關注
0文章
87瀏覽量
11472 -
Arduino
+關注
關注
185文章
6434瀏覽量
185199 -
樹莓派
+關注
關注
113文章
1649瀏覽量
104844 -
MQTT
+關注
關注
5文章
550瀏覽量
22092 -
ESP32
+關注
關注
13文章
900瀏覽量
16017
原文標題：ESP32 運行MQTT客戶端進行主題的發布和訂閱
文章出處：【微信號：凌順實驗室，微信公眾號：凌順實驗室】歡迎添加關注！文章轉載請注明出處。
發布評論請先 登錄
相關推薦
MQTT協議介紹之三：發送和接收消息
強烈建議您先閱讀。上一節我們研究了建立MQTT客戶端和代理之間的連接，所以本周我們將討論發送和接收消息。發布MQTT客戶端連接到代理后，可以
發表于 08-25 19:54
【HarmonyOS】 MQTT 協議開發入門
：發布者（Publish）、代理（Broker）（服務器）、訂閱者（Subscribe）。其中，消息的發布者和訂閱者都是客戶端，消息代理是服
發表于 11-03 17:38
MQTT 協議開發入門
者（Publish）、代理（Broker）（服務器）、訂閱者（Subscribe）。其中，消息的發布者和訂閱者都是客戶端，消息代理是服務器，消息發布
發表于 11-05 09:38
基于Eclipse paho-mqtt源碼包的MQTT客戶端設計（上）
之間，它可以：接受來自客戶的網絡連接接收客戶發布的應用信息處理來自客戶端的訂閱和退訂請求向訂閱的
發表于 08-04 16:28
基于Eclipse paho-mqtt源碼包的MQTT客戶端設計（下）
，如果有訂閱的 Topic 沒有設置回調函數，則使用該默認回調函數啟動 MQTT 客戶端配置完成 MQTT 客戶端實例后，需要啟動
發表于 08-04 16:33
基于socket API開發一個高性能高穩定性的跨平臺MQTT客戶端
方是API函數接口，實現了客戶端的申請、釋放、設置參數、連接服務器、斷開連接、訂閱主題、取消訂閱主題、發
發表于 08-09 11:15
OpenHarmony開發之MQTT講解
者也可以同時作為信息訂閱者。如下圖所示：上圖中的所有客戶端都是圍繞“空調溫度”這一主題進行通訊的。對于“空調溫度”這一主題，手機和電腦
發表于 11-16 12:05
esp32-c3 mqtt遺囑實驗沒成功的原因？
;/test" topic 的消息 消息內容cono 好像去掉了后面一部分。后來改變esp32-c3（發布者）的內容都還是訂閱的客戶端這個topic出現。
發表于 03-07 09:00
如何在rtos mqtt示例中獲取客戶端句柄？
中，我調用
代碼：全選esp_mqtt_client_publish()
但我沒有客戶端句柄，它只在回調中設置。我嘗試使用客戶端句柄的全局副本，但它沒有用，除了是個壞主意：
代碼：全選
發表于 05-15 08:13
工商網監
評論