目的

MQTT是比較常用在物聯(lián)網(wǎng)設備中的通訊協(xié)議，這篇文章將使用 Arudino ESP32 作為MQTT客戶端進行通訊使用演示。目前Arduino的MQTT客戶端庫中最常使用的是 PubSubClient ，所以本文也將以此進行說明。

主頁：https://pubsubclient.knolleary.net/
項目地址：https://github.com/knolleary/pubsubclient
目前 PubSubClient 庫版本為 v2.8 ，主要基于 MQTT 3.1.1 ，不支持 MQTT 5.0 的新增特性，訂閱主題只支持 Qos 0 和 1。

This library provides a client for doing simple publish/subscribe messaging with a server that supports MQTT.

基礎說明

MQTT的一些基礎內(nèi)容可以參考下面文章：
《MQTT基礎入門與資料收集》
因為測試需要有 MQTT Broker（服務器） ，可以參考上面文章進行啟動，或者也可以申請一個免費的在線的云服務使用。

PubSubClient 庫使用很簡單，主要就是分為下面幾步：

1. 聲明 PubSubClient 對象；
2. 因為MQTT需要在TCP之上工作，所以需要給 PubSubClient 對象一個TCP對象；
3. 使用 setServer 方法設置MQTT服務器的地址和端口號；
4. 使用 setCallback 方法設置通訊消息回調(diào)函數(shù) void callback(char *topic, byte *payload, unsigned int length)（如果不需要訂閱消息則無需此步驟）；
5. 使用 connect 方法啟動連接；
6. 使用 subscribe 方法訂閱主題或使用 publish 方法向某個主題發(fā)布消息；

上面步驟中 2、3、4 步順序并無要求，并且可以在第一步聲明對象的構(gòu)造函數(shù)中直接傳入。

在使用 connect 方法進行連接時可以選擇填入 Will Qos Will Retain Will Message cleanSession 信息（默認為 0 0 0 1 ）。發(fā)送消息時可以選擇填入 retained （默認為 false ）。

默認情況下發(fā)送和接收數(shù)據(jù)都會依賴buffer，當發(fā)送或者接收的消息比buffer可容納的空間（默認256字節(jié)）大的時候?qū)雎赃@條消息?？梢允褂?setBufferSize 方法來設置buffer大小。

上面的 publish 方法發(fā)送消息時會先將消息拷貝到緩存，這在大數(shù)據(jù)發(fā)送時效率并不好，可以先使用 beginPublish 方法啟動傳輸，然后單次或多次使用 write 方法寫數(shù)據(jù)（也可以使用 print 等方法），最后使用 endPublish 完成本次消息發(fā)送，減少一次拷貝，效率上會高很多。

可以使用 setKeepAlive 方法來設置 Keep Alive 時間（默認為15s）。

使用 disconnect 方法可以關(guān)閉連接，使用 connected 方法可以檢查是否連接。使用 unsubscribe 方法可以取消訂閱消息。

使用 state 方法可以獲得 PubSubClient 對象當前的狀態(tài)，狀態(tài)定義如下：

// Possible values for client.state()
#define MQTT_CONNECTION_TIMEOUT     -4
#define MQTT_CONNECTION_LOST        -3
#define MQTT_CONNECT_FAILED         -2
#define MQTT_DISCONNECTED           -1
#define MQTT_CONNECTED               0
#define MQTT_CONNECT_BAD_PROTOCOL    1
#define MQTT_CONNECT_BAD_CLIENT_ID   2
#define MQTT_CONNECT_UNAVAILABLE     3
#define MQTT_CONNECT_BAD_CREDENTIALS 4
#define MQTT_CONNECT_UNAUTHORIZED    5

示例代碼

#include <WiFi.h>
#include <PubSubClient.h>// WiFi相關(guān)配置信息
const char *wifi_ssid = "********";
const char *wifi_password = "********";// MQTT相關(guān)配置信息
const char *mqtt_broker_addr = "********"; // 服務器地址
const uint16_t mqtt_broker_port = 1883; // 服務端口號            
const char *mqtt_username = "********"; // 賬號（非必須）
const char *mqtt_password = "********"; // 密碼（非必須）
const uint16_t mqtt_client_buff_size = 4096; // 客戶端緩存大小（非必須）
String mqtt_client_id = "esp32_client"; // 客戶端ID
const char *mqtt_topic_pub = "esp32/test"; // 需要發(fā)布到的主題
const char *mqtt_topic_sub = "esp32/test"; // 需要訂閱的主題WiFiClient tcpClient;
PubSubClient mqttClient;// MQTT消息回調(diào)函數(shù)，該函數(shù)會在PubSubClient對象的loop方法中被調(diào)用
void mqtt_callback(char *topic, byte *payload, unsigned int length)
{Serial.printf("Message arrived in topic %s, length %d\n", topic, length);Serial.print("Message:");for (int i = 0; i < length; i++){Serial.print((char)payload[i]);}Serial.println("\n----------------END----------------");
}void setup()
{Serial.begin(115200);Serial.println();// 連接網(wǎng)絡Serial.printf("\nConnecting to %s", wifi_ssid);WiFi.begin(wifi_ssid, wifi_password);while (WiFi.status() != WL_CONNECTED){delay(500);Serial.print(".");}Serial.println("ok.");Serial.print("IP address: ");Serial.println(WiFi.localIP());// 設置MQTT客戶端mqttClient.setClient(tcpClient);mqttClient.setServer(mqtt_broker_addr, mqtt_broker_port);mqttClient.setBufferSize(mqtt_client_buff_size);mqttClient.setCallback(mqtt_callback);
}unsigned long previousConnectMillis = 0; // 毫秒時間記錄
const long intervalConnectMillis = 5000; // 時間間隔
unsigned long previousPublishMillis = 0; // 毫秒時間記錄
const long intervalPublishMillis = 5000; // 時間間隔void loop()
{unsigned long currentMillis = millis(); // 讀取當前時間// 連接MQTT服務器if (!mqttClient.connected()) // 如果未連接{if (currentMillis - previousConnectMillis > intervalConnectMillis){previousConnectMillis = currentMillis;mqtt_client_id += String(WiFi.macAddress()); // 每個客戶端需要有唯一的ID，不然上線時會把其他相同ID的客戶端踢下線if (mqttClient.connect(mqtt_client_id.c_str())) // 嘗試連接服務器// if (mqttClient.connect(mqtt_client_id.c_str(), mqtt_username, mqtt_password)){mqttClient.publish(mqtt_topic_pub, "hello mqtt!"); // 連接成功后可以發(fā)送消息mqttClient.subscribe(mqtt_topic_sub); // 連接成功后可以訂閱主題}}}// 定期發(fā)送消息if (mqttClient.connected()){if (currentMillis - previousPublishMillis >= intervalPublishMillis) // 如果和前次時間大于等于時間間隔{previousPublishMillis = currentMillis;mqttClient.publish(mqtt_topic_pub, "naisu 233~~~");}}// 處理MQTT事務mqttClient.loop();
}

上面代碼示例演示的是非加密的mqtt，實際業(yè)務中更多的可能會使用加密的mqtts，這個時候TCP客戶端就需要使用 #include <WiFiClientSecure.h> 庫中的 WiFiClientSecure 對象了。TCP客戶端需要使用 setCACert 或者 getFingerprintSHA256 等方法設置證書或者指紋，另外可能需要從NTP服務器獲取時間。（ WiFiClientSecure對象可以使用 setInsecure 方法，可以不用管證書這些，測試使用沒問題，實際使用中可能會有安全風險）

總結(jié)

MQTT作為客戶端使用本身比較簡單，PubSubClient用起來也非常簡單，基本上一般的使用有上面內(nèi)容就夠了。