ESP32 MQTT的庫有很多，凌順實驗室（lingshunlab.com）這次主要使用AsyncMQTT_ESP32，以后有機(jī)會再更多的MQTT其他庫的使用方法。

前提條件

樹莓派部署本地的MQTT服務(wù)端，具體安裝請查看以下連接：

https://lingshunlab.com/book/raspberry-pi/raspberry-pi-install-mosquitto-mqtt-server-and-test-mqtt

ESP32和樹莓派在同一WIFI網(wǎng)絡(luò)里面

效果實現(xiàn)

凌順實驗室（lingshunlab.com）在本示例展示了使用兩個ESP32，分別實現(xiàn)發(fā)布MQTT的主題消息和訂閱并輸出MQTT的主題內(nèi)容。當(dāng)然，可能會問能不能一個ESP32同時又是發(fā)布者，又是訂閱者？答案是可以的，因為作為客戶端，都是對中間商做信息交換。

BOM

需要準(zhǔn)備2個ESP32，
一個ESP32用于發(fā)布，
一個ESP32用于訂閱。

庫的安裝

可以在Arduino IDE的庫管理里搜索并安裝：

點擊菜單欄的「工具」---> 「庫管理」，然后在搜索框中輸入“AsyncMQTT_ESP32”，點擊安裝即可

下圖在我本地已經(jīng)安裝好了：

又或者在Github中下載，并安裝到Arduino的 "libraries"文件夾里

Github 地址：

https://github.com/khoih-prog/AsyncMQTT_ESP32

程序提點

1， 首先，需要加載AsyncMQTT_ESP32的庫

#include 

2，配置MQTT的服務(wù)器信息，可以是IP或者域名的方式

//#define MQTT_HOST         IPAddress(192, 168, 100, 100)
#define MQTT_HOST         "broker.emqx.io"        // Broker address
#define MQTT_PORT         1883

3，設(shè)置主題，發(fā)布需要主題，訂閱也需要主題

const char *Topic  = "lingshunlab/ESP32";               // 主題

4，創(chuàng)建MQTT客戶端的實例

// 創(chuàng)建MQTT客戶端的實例，名為mqttClient
AsyncMqttClient mqttClient;

5，認(rèn)識mqttClient的可用的回調(diào)函數(shù)

當(dāng)MQTT觸發(fā)特定事件的時候，可以配置自定義的函數(shù)

  mqttClient.onConnect(onMqttConnect);  // 設(shè)置 當(dāng)MQTT連接時的回調(diào)函數(shù)
  mqttClient.onDisconnect(onMqttDisconnect); // 設(shè)置 當(dāng)MQTT斷開連接時的回調(diào)函數(shù)
  mqttClient.onSubscribe(onMqttSubscribe); // 設(shè)置 當(dāng)MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.onUnsubscribe(onMqttUnsubscribe); // 設(shè)置 當(dāng)MQTT取消訂閱主題時的回調(diào)函數(shù)
  mqttClient.onMessage(onMqttMessage); // 設(shè)置 當(dāng)MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.onPublish(onMqttPublish); // 設(shè)置 當(dāng)取消MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設(shè)置 MQTT服務(wù)器信息

6， 連接MQTT服務(wù)器

  mqttClient.setServer(MQTT_HOST, MQTT_PORT); //連接MQTT服務(wù)器

7，發(fā)布主題

通過以下代碼，可以對配置好的主題發(fā)布消息

// 發(fā)布主題消息
  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設(shè)置為2
  uint16_t packetIdSub = mqttClient.subscribe(SubTopic, 2);
  Serial.print("Subscribing at QoS 2, packetId: ");
  Serial.println(packetIdSub);

9，當(dāng)發(fā)生主題消息變化的時候的回調(diào)函數(shù)

mqttClient的回調(diào)函數(shù)有很多種，請仔細(xì)學(xué)習(xí)查看例子中其他的回調(diào)函數(shù)。在這里，特別說明一下onMessage的回調(diào)函數(shù)onMqttMessage（這個函數(shù)名稱你可以自己定義，隨喜），里面有不少參數(shù)，例如topic，payload等，其中payload即是消息的內(nèi)容，可以通過輸出顯示。

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);   // 輸出消息內(nèi)容
}

10，請查看AsyncMQTT_ESP32的官方例子

可以學(xué)習(xí)到FreeRTOS的多線程如何應(yīng)用。

程序代碼

發(fā)布主題de完整代碼

#include 

// 配置 WIFI 
#define WIFI_SSID         "***your wifi***"
#define WIFI_PASSWORD     "***your wifi password***"

// 加載AsyncMQTT_ESP32庫
#include  

// 配置MQTT服務(wù)器地址和端口
#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";               // 發(fā)布消息的主題

AsyncMqttClient mqttClient; // 創(chuàng)建 MQTT客戶端實例

void onMqttConnect(bool sessionPresent)   // 編寫對應(yīng)的回調(diào)函數(shù)
{
  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);   // 輸出消息內(nèi)容
}

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);  // 設(shè)置 當(dāng)MQTT連接時的回調(diào)函數(shù)
  mqttClient.onDisconnect(onMqttDisconnect); // 設(shè)置 當(dāng)MQTT斷開連接時的回調(diào)函數(shù)
  mqttClient.onMessage(onMqttMessage); // 設(shè)置 當(dāng)MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.onPublish(onMqttPublish); // 設(shè)置 當(dāng)取消MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設(shè)置 MQTT服務(wù)器信息
  mqttClient.connect(); // 連接 MQTT

  delay(500);
}

void loop()
{
  // 發(fā)布主題消息
  uint16_t packetIdPub = mqttClient.publish(PubTopic, 2, true, "welcome to Lingshunlab.com");
  Serial.print("Publisshing at QoS 2, packetId: ");
  Serial.println(packetIdPub);
  delay(2000);
}

上傳代碼后，程序?qū)冗B接WIFI，然后連接MQTT服務(wù)器，再之后每隔2秒發(fā)布一個對應(yīng)主題的消息

訂閱主題de完整代碼

#include 

// 配置 WIFI 
#define WIFI_SSID         "***your wifi***"
#define WIFI_PASSWORD     "***your wifi password***"

// 加載 AsyncMQTT_ESP32 庫
#include 

// 配置MQTT服務(wù)器地址和端口
#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設(shè)置為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); // 設(shè)置 當(dāng)MQTT連接時的回調(diào)函數(shù)
  mqttClient.onDisconnect(onMqttDisconnect);  // 設(shè)置 當(dāng)MQTT斷開連接時的回調(diào)函數(shù)
  mqttClient.onSubscribe(onMqttSubscribe); // 設(shè)置 當(dāng)MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.onUnsubscribe(onMqttUnsubscribe); // 設(shè)置 當(dāng)取消MQTT訂閱主題時的回調(diào)函數(shù)
  mqttClient.onMessage(onMqttMessage); // 設(shè)置 當(dāng)MQTT收到主題消息時的回調(diào)函數(shù)
  mqttClient.setServer(MQTT_HOST, MQTT_PORT); // 設(shè)置 MQTT服務(wù)器信息
  mqttClient.connect(); // 連接 MQTT

  delay(500);
}

void loop()
{

}

上傳代碼后，程序?qū)冗B接WIFI，然后連接MQTT服務(wù)器，當(dāng)連接MQTT時，則會訂閱主題，之后每隔2秒就會收到主題發(fā)布的消息

審核編輯：劉清