MQTT是一種用于通過 Internet 發送和接收消息的協議。我們之前在物聯網電表和Raspberry Pi Alexa中使用此協議在互聯網上發布數據。在本教程中，我們將更多地了解 MQTT 及其相關術語。在這里，我們將使用Raspberry Pi 作為本地 MQTT 代理，并通過 MQTT 應用程序儀表板控制連接到 NodeMCU ESP12E 的 LED。DHT11 傳感器也連接到 NodeMCU，因此我們再次使用 Raspberry Pi 作為本地 MQTT 代理，在 MQTT 儀表板上讀取溫度和濕度。

因此，讓我們從了解 MQTT 及其相關術語開始。

什么是 MQTT？

MQTT 代表IBM 設計的 Message Queue Telemetry Transport。該協議簡單輕量，用于通過互聯網發送和接收消息，專為低帶寬使用的設備而設計。如今，該協議經常在物聯網設備中用于發送和接收傳感器數據。此外，在基于物聯網的家庭自動化系統中，無需使用大量互聯網數據即可輕松使用此協議。

在 MQTT 中經常使用的術語很少：

訂閱和發布

信息

話題

經紀人

1.訂閱和發布：訂閱是從其他設備獲取數據，發布是向其他設備發送數據。

當 device1 將數據發送到 device2 時，它稱為發布者，另一個稱為訂閱者，反之亦然。

2. 消息：消息是我們發送和接收的信息。它可以是數據或任何類型的命令。例如，如果我們將溫度數據發布到云端，則此溫度數據稱為消息。

3. 主題：這是您注冊對傳入消息感興趣的方式或您如何指定要發布消息的位置。主題用正斜杠分隔的字符串表示。使用 MQTT 在主題上發布數據，然后 MQTT 設備訂閱主題以獲取數據。

4. MQTT Broker：這個東西負責接收所有發布者的消息，過濾消息，然后將消息發布給感興趣的訂閱者。

當這個代理托管在云上時，它被稱為 MQTT 云。有許多基于云的 MQTT 服務，例如 Adafruit IO、MQTT.IO、IBM bluemix、Microsoft Azure 等。MQTT 也可以與流行的 Amazon AWS 云一起使用，我們在Amazon AWS 入門教程中對此進行了說明。

我們可以使用 Raspberry Pi 創建自己的 MQTT 代理。這將是本地 MQTT 代理，即您只能在本地網絡上發送和接收數據，而不是從任何地方。所以在這里我們將在 Raspberry Pi 中安裝 Mosquitto MQTT 代理，使其成為本地 MQTT 代理，并將溫度數據從 NodeMCU 發送到 MQTT 儀表板應用程序。此外，我們將使用代理控制連接到 NodeMCU 的 LED。

在樹莓派上安裝 Mosquitto MQTT 代理

在您的 Raspberry pi 中打開終端并鍵入以下命令以安裝代理

sudo apt update 
sudo apt install -y mosquitto mosquitto-clients

等待安裝完成。現在，要在啟動樹莓派時啟動代理，請鍵入以下命令

sudo systemctl 啟用 mosquitto.service

就是這樣，我們都準備好啟動我們的 MQTT 代理了。要檢查它是否安裝正確，請輸入以下命令

蚊子-v

此命令將為您提供 MQTT 代理的版本。它應該是 1.4.x 或更高版本。

測試 Raspberry Pi Mosquitto Broker

1. 使用以下命令在后臺運行 Mosquitto 代理

蚊子-d

2. 現在，我們將使用以下命令在 exampleTopic 中訂閱一個主題

mosquitto_sub -d -t exampleTopic

3. 現在，我們將發布一些消息到 exampleTopic

mosquitto_pub -d -t exampleTopic -m "Hello world!"

您將收到 Hello World！用戶終端中的消息。

現在，在我們使用 NodeMCU 和MQTT 儀表板應用程序的案例中，是時候控制和從另一臺設備獲取數據了。

首先，我們將通過使用 App 發送命令來控制 LED，因此在這種情況下，NodeMCU 表現為訂閱者，而 App 表現為發布者。

然后 ESP12E 還連接了 DHT11 傳感器，并將此溫度讀數發送到 Mobile MQTT 應用程序，因此在這種情況下，移動設備將成為訂閱者，NodeMCU 將成為發布者。為了在各自的主題上轉發這些消息，使用了 Mosquitto MQTT 代理。

電路原理圖

如圖所示連接電路。這里 DHT11 用于溫度讀數，但也可以使用LM35 溫度傳感器。

讓我們開始編寫 NodeMCU 的代碼來訂閱和發布數據。

代碼和解釋

在這里，我們將使用 Adafruit MQTT 庫模板，并將更改代碼中所需的內容。只需更改一些內容，即可使用相同的代碼將數據發布和訂閱到 Adafruit IO 云。

為此，從Sketch -》 Include Library -》 Manage Libraries 下載 Adafruit MQTT 庫。搜索 Adafruit MQTT 并安裝它。安裝庫后。轉到 示例-》 Adafruit mqtt 庫-》 mqtt_esp8266

然后根據我們的 Raspberry Pi IP 地址和 Wi-Fi 憑據編輯此代碼。

包括 ESP8266WIFI 和 Adafruit MQTT 的所有庫。

#include  
#include "Adafruit_MQTT.h" 
#include "Adafruit_MQTT_Client.h" 
#include "DHT.h"

然后為您想要連接 ESP-12e 的 Wi-Fi 定義 SSID 和密碼。確保您的 RPi 和 NodeMCU 連接到同一個網絡。

#define WLAN_SSID "xxxxxxxx" 
#define WLAN_PASS "xxxxxxxxxxx"

本節定義 Adafruit 服務器，在本例中是您的 Raspberry Pi 的 IP 地址和服務器端口。

#define AIO_SERVER "你的 Pi 的 IP 地址" 
#define AIO_SERVERPORT 1883                  

下面的字段將保持空白，因為我們沒有使用 Adafruit 云。

#define AIO_USERNAME "" 
#define AIO_KEY ""

然后創建一個 ESP8266 WiFiClient 類來連接到 MQTT 服務器。

WiFiClient客戶端；

通過傳入 WiFi 客戶端和 MQTT 服務器以及登錄詳細信息來設置 MQTT 客戶端類。

Adafruit_MQTT_Client mqtt(&client, AIO_SERVER, AIO_SERVERPORT, AIO_USERNAME, AIO_KEY);

設置一個名為“溫度”的提要來發布溫度。

Adafruit_MQTT_Publish 溫度 = Adafruit_MQTT_Publish(&mqtt, AIO_USERNAME "/feeds/temperature");

設置一個名為“led1”的提要來訂閱更改。

Adafruit_MQTT_Subscribe led1 = Adafruit_MQTT_Subscribe(&mqtt, AIO_USERNAME "/feeds/led");

在setup函數中，我們聲明了要在其上獲得輸出的 NodeMCU 的 PIN。然后，將 NodeMCU 連接到 Wi-fi 接入點。

無效設置（） { 
  Serial.begin（115200）；
  延遲（10）；
  pinMode（LED，輸出）；
  Serial.println(F("Adafruit MQTT demo")); 
  // 連接到 WiFi 接入點。
  序列號.println(); 序列號.println(); 
  Serial.print("正在連接"); 
  Serial.println(WLAN_SSID); 
  WiFi.開始（WLAN_SSID，WLAN_PASS）；
  而 (WiFi.status() != WL_CONNECTED) { 
.... 
…… 
…
為 led 提要設置 MQTT 訂閱。
  mqtt.subscribe(&led1); 
}

在循環函數中，我們將使用 MQTT_connect() 確保與 MQTT 服務器的連接處于活動狀態；功能。

無效循環() { 
  MQTT_connect();

現在，訂閱 'led' 提要并從 MQTT 代理獲取字符串，并使用atoi() 將此字符串轉換為數字；函數并使用digitalWrite()將此數字寫入 LED 引腳；功能。

Adafruit_MQTT_Subscribe *訂閱；

  while ((subscription = mqtt.readSubscription(20000))) { 
    if (subscription == &led1) { 
      Serial.print(F("Got: ")); 
      Serial.println((char *)led1.lastread); 
      int led1_State = atoi((char *)led1.lastread); 
      數字寫入（LED，led1_State）；
    }

現在，獲取變量中的溫度并使用Temperature.publish(t)函數發布該值。

浮動 t = dht.readTemperature(); 
… 
..

  if (!Temperature.publish(t)) { 
    Serial.println(F("Failed")); 
  } else { 
    Serial.println(F("OK!")); 
  }

#include

#include "Adafruit_MQTT.h"

#include "Adafruit_MQTT_Client.h"

#include "DHT.h"


#define LED D5

#define DHTPIN D4


#define WLAN_SSID "awesome"

#define WLAN_PASS "awesome12"


#define AIO_SERVER "192.168.43.177" //RPi的IP地址

#define AIO_SERVERPORT 1883

#define AIO_USERNAME ""

#define AIO_KEY ""


WiFiClient客戶端；

Adafruit_MQTT_Client mqtt(&client, AIO_SERVER, AIO_SERVERPORT, AIO_USERNAME, AIO_KEY);

Adafruit_MQTT_Publish 溫度 = Adafruit_MQTT_Publish(&mqtt, AIO_USERNAME "/feeds/temperature");

Adafruit_MQTT_Subscribe led1 = Adafruit_MQTT_Subscribe(&mqtt, AIO_USERNAME "/feeds/led");

#define DHTTYPE DHT11 // DHT 11


DHT dht(DHTPIN, DHTTYPE);

uint32_t 延遲時間；



/****************************** 草圖代碼 ************************ ******************/


無效 MQTT_connect();


無效設置（） {

Serial.begin（115200）；


延遲（10）；

pinMode（LED，輸出）；

Serial.println(F("Adafruit MQTT demo"));


// 連接到 WiFi 接入點。

序列號.println(); 序列號.println();

Serial.print("正在連接");

Serial.println(WLAN_SSID);


WiFi.開始（WLAN_SSID，WLAN_PASS）；

而（WiFi.status（）！= WL_CONNECTED）{

延遲（500）；

Serial.print(".");

}

Serial.println();


Serial.println("WiFi 連接");

Serial.println("IP地址："); Serial.println(WiFi.localIP());



//設置 DHT 傳感器

dht.begin();


// 為 onoff feed 設置 MQTT 訂閱。

mqtt.subscribe(&led1);



}


uint32_t x = 0;


void loop() {

// 確保與 MQTT 服務器的連接處于活動狀態（這將建立第一個

// 連接并在斷開連接時自動重新連接）。請參閱下面的 MQTT_connect

// 函數定義。

MQTT_connect();

// 這是我們的“等待收到的訂閱數據包”繁忙的子循環

// 試著把你的時間花在這里


Adafruit_MQTT_Subscribe *訂閱；

while ((subscription = mqtt.readSubscription(20000))) {

if (subscription == &led1) {

Serial.print(F("Got: "));

Serial.println((char *)led1.lastread);

int led1_State = atoi((char *)led1.lastread);

數字寫入（LED，led1_State）；

Serial.println("onnn");

}



}


// 以攝氏度讀取溫度（默認）

float t = dht.readTemperature();

// 檢查是否有任何讀取失敗并提前退出（重試）。

if (isnan(t)) {

Serial.println("讀取 DHT 傳感器失敗！");

返回;

}

if (! Temperature.publish(t)) {

Serial.println(F("Failed"));

} else {

Serial.println(F("OK!"));

}


}


// 根據需要連接和重新連接到 MQTT 服務器的函數。

// 應該在循環函數中調用，如果連接會小心。

無效 MQTT_connect() {

int8_t ret;


// 如果已經連接則停止。

if (mqtt.connected()) {

返回;

}


Serial.print("正在連接 MQTT...");

uint8_t 重試次數 = 3；

while ((ret = mqtt.connect()) != 0) { // connect 將返回 0 連接

Serial.println(mqtt.connectErrorString(ret));

Serial.println("5 秒后重試 MQTT 連接...");

mqtt.disconnect();

延遲（5000）；// 等待 5 秒

重試--;

if (retries == 0) {

// 基本上死了，等待 WDT 重置我

while (1);

}

}

Serial.println("MQTT 已連接！");

}