通过 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(https://cloud.emqx.io)注册地址注册
• 注册后点击 EMQX Cloud 申请免费试用15天部署
  

• 查看 broker 连接地址
  

代码编写[mw_shl_code=c,false]#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);
        }
    }
}[/mw_shl_code]

按照以下操作编辑代码以适合您自己的 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 消息。
    

    
    
    • 创建 MQTT 连接
    • 订阅主题，接受测试消息
      
  • 使用Python MQTT 客户端查看上报消息
    [mw_shl_code=python,false]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()[/mw_shl_code]Python 脚本运行截图：
    
    
• 故障排除：为了执行故障排除，然后将 USB 适配器与 PC 连接并在 Arduino IDE 中选择 USB-TTL 适配器的端口。打开“串行监视器”以查看由串行输出产生的调试信息
  

总结

至此为止，完成了从 NodeMCU 采集数据，并上传到 EMQ 提供的 MQTT 云服务，最后由 Python 写的后端程序对数据进行处理的简单过程。但在实际的生产应用中，会需要更高的要求，比如，

• 更加安全的连接方式
• 对物联网数据进行实时处理
• 对数据进行持久化
• 更大规模的连接要求
  

EMQ 企业版(https://www.emqx.io/products/enterprise)，及其云服务在解决上述问题已经提供了很好的解决方案，有兴趣的读者可以参考相关链接了解更多的信息。

为了实现数据的高安全性（避免上传到云端），降低业务处理时延，以及数据传输成本，在解决方案中可以考虑采用边缘计算。Azure IoT Edge 和 AWS 的 Greengrass 提供了在边缘端的解决方案。EMQ 也提供了开源的超轻量级边缘物联网实时数据分析 (IoT Edge streaming analytics) 方案 Kuiper，读者可以参考这篇文章以获取更详细的信息。