通過基於 NodeMCU (ESP8266) 將傳感器數據上傳至 MQTT 雲服務

簡介

本示例將演示如何通過 基於 ESP8266 的 NodeMCU，把通過 DHT11 傳感器搜集到的溫、濕度數據通過 MQTT 協議將其上報到雲端的 MQTT 服務，並且展示應用端如何訂閱到這些數據並對之進行處理的過程。本文使用 MQTT 協議的原因在於該協議比較輕量級，節約能耗，非常適合於物聯網的相關使用場景；目前各大公有雲雲提供商基本上都開放了基於 MQTT 協議的 IoT Hub 服務。比如 AWS 的 IoT Core，以及 Azure 的 IoT Hub 等，通過 MQTT 協議可以非常方便的將這些數據直接接入這些公有雲服務。

本示例的總體架構如下

配置

硬件配置

• NodeMCU board x 1：NodeMCU 是一個開源的 IoT （硬件）開發平台，NodeMCU 包含了可以運行在 ESP8266 Wi-Fi SoC芯片之上的固件,以及基於 ESP-12 模組的硬件。“NodeMCU” 缺省一般指的是固件，而不是開發套件。固件使用 Lua 腳本語言。
• DHT11 temperature/humidity sensor x 1：DHT11 數字溫濕度傳感器是一款含有已校准數字信號輸出的溫濕度復合傳感器
• 面包板（Breadboard ）x 1
• 跳線（Jumper wires）若干
• 連接圖（Connection Graph）請參考如下截圖

Arduino 配置

• 下載並安裝 CH340G USB 驅動
• 安裝 ESP8266模塊
• 安裝 PubSubClient 庫 (by Nick O'Leary)
  Sketch -> Include Library -> Manage Libraries... -> Type PubSub in Search field -> Install

MQTT 雲服務

通過 NodeMCU 將數據采集成功后，需要將其發送到雲端的 MQTT 雲服務。本文使用由 EMQX 提供的 MQTT 雲服務, 讀者也可以根據自己的情況來選擇別的 MQTT 雲服務，比如 Azure IoT Hub 或者 AWS IoT Core，每種雲服務在接入的時候需要提供不同的認證方式，因此通過 NodeMCU 連接雲端的 MQTT 服務等時候需要根據目標雲服務的安全要求設定連接方式。本文為了簡單起見，使用了非安全的連接方式，在正式生產環境中必須設定安全認證方式的連接。

• 點擊 EMQX Cloud 注冊地址注冊
• 注冊后點擊 EMQX Cloud 申請免費試用15天部署

• 查看 broker 連接地址

代碼編寫

#include <ESP8266WiFi.h>

#include <PubSubClient.h>

#include "DHT.h"

#define DHTPIN D4     // what pin we're connected to
#define wifi_ssid "xxxxx"
#define wifi_password "xxxxx"

#define mqtt_server "broker-internet-facing-f1429d8cb54ca4a7.elb.us-east-1.amazonaws.com"  // MQTT Cloud address
#define humidity_topic "humidity"
#define temperature_topic "temperature"

#define DHTTYPE DHT11   // DHT 11

WiFiClient espClient;
PubSubClient client(espClient);
DHT dht(DHTPIN, DHTTYPE);

void setup() {
    Serial.begin(115200);
    setup_wifi();
    client.setServer(mqtt_server, 1883);
    dht.begin();
}

void setup_wifi() {
    delay(10);
    WiFi.begin(wifi_ssid, wifi_password);
    while (WiFi.status() != WL_CONNECTED) {
        delay(500);
        Serial.print(".");
    }
}

void reconnect() {
    // Loop until we're reconnected
    while (!client.connected()) {
        Serial.print("Attempting MQTT connection...");
        if (client.connect("nodeMcuDHT11")) {
            Serial.println("connected");
        } else {
            Serial.print("failed, rc=");
            Serial.print(client.state());
            Serial.println(" try again in 5 seconds");
            delay(5000);
        }
    }
}

bool checkBound(float newValue, float prevValue, float maxDiff) {
    return newValue < prevValue - maxDiff || newValue > prevValue + maxDiff;
}

long lastMsg = 0;
float temp = 0.0;
float hum = 0.0;
float diff = 1.0;

void loop() {
    if (!client.connected()) {
        reconnect();
    }
    client.loop();

    long now = millis();
    if (now - lastMsg > 30000) {
        // Wait a few seconds between measurements
        lastMsg = now;

        float newTemp = dht.readTemperature();
        float newHum = dht.readHumidity();
        if (checkBound(newTemp, temp, diff)) {
            temp = newTemp;
            Serial.print("New temperature:");
            Serial.println(String(temp).c_str());
            client.publish(temperature_topic, String(temp).c_str(), true);
        }

        if (checkBound(newHum, hum, diff)) {
            hum = newHum;
            Serial.print("New humidity:");
            Serial.println(String(hum).c_str());
            client.publish(humidity_topic, String(hum).c_str(), true);
        }
    }
}

按照以下操作編輯代碼以適合您自己的 Wi-Fi 和 MQTT 設置

• Wi-Fi 設置

  #define wifi_ssid ""
  #define wifi_password ""
  

• Broker 服務器設置

  #define mqtt_server "broker-internet-facing-f1429d8cb54ca4a7.elb.us-east-1.amazonaws.com"
  

• Arduion 配置

運行

• 代碼上傳

  將 NodeMCU 通過 USB 連接到 PC 並在 Arduion IDE 中選擇 115200 端口，使用 upload 按鈕編譯草圖並將其上傳到設備

• 打開 Arduino monitor window 查看數據上報

• MQTT 客戶端接受消息

  • 使用 MQTT Websocket Toolkit 測試上報消息

    MQTT Websocket Toolkit 是 EMQ 最近開源的一款 MQTT (WebSocket) 測試工具，支持線上 (tools.emqx.io) 訪問使用, 我們可以方便的用於驗證 NodeMCU 是否上報 MQTT 消息。

    1. 創建 MQTT 連接
       
    2. 訂閱主題，接受測試消息
       

  • 使用Python MQTT 客戶端查看上報消息

    from paho.mqtt import client as mqtt
    
    
    def on_connect(client, userdata, flags, rc):
        # connect mqtt broker
        client.subscribe([("temperature", 0), ("humidity", 0)])
    
    
    def on_message(client, userdata, msg):
        # sub dht11 temperature/humidity data
        print(f"{msg.topic}: {msg.payload.decode()}")
    
    
    def run():
        client = mqtt.Client()
        # Edit MQTT Cloud address
        client.connect("broker-internet-facing-f1429d8cb54ca4a7.elb.us-east-1.amazonaws.com", 1883)
        client.on_connect = on_connect
        client.on_message = on_message
        client.loop_forever()
    
    
    if __name__ == '__main__':
        run()
    

    Python 腳本運行截圖：

  

• 故障排除：為了執行故障排除，然后將 USB 適配器與 PC 連接並在 Arduino IDE 中選擇 USB-TTL 適配器的端口。打開“串行監視器”以查看由串行輸出產生的調試信息

總結

至此為止，完成了從 NodeMCU 采集數據，並上傳到 EMQ 提供的 MQTT 雲服務，最后由 Python 寫的后端程序對數據進行處理的簡單過程。但在實際的生產應用中，會需要更高的要求，比如，

• 更加安全的連接方式
• 對物聯網數據進行實時處理
• 對數據進行持久化
• 更大規模的連接要求

EMQ 企業版，及其雲服務在解決上述問題已經提供了很好的解決方案，有興趣的讀者可以參考相關鏈接了解更多的信息。

為了實現數據的高安全性（避免上傳到雲端），降低業務處理時延，以及數據傳輸成本，在解決方案中可以考慮采用邊緣計算。Azure IoT Edge 和 AWS 的 Greengrass 提供了在邊緣端的解決方案。EMQ 也提供了開源的超輕量級邊緣物聯網實時數據分析 (IoT Edge streaming analytics) 方案 Kuiper，讀者可以參考這篇文章以獲取更詳細的信息。