前言

注：本文系湛江市第十七中學星火創客團隊及嶺南師范學院物聯網俱樂部原創部分參賽專案，轉載請保留宣告，

一般在做物聯網相關專案的時候，物聯網與軟體對接的時候，需要先采集資料，將資料上傳至云端后，再將資料轉存到MySQL中，方便后臺人員去調取資料，本文采用的方法是一種使用Wemos（ESP8266）將采集到的資料通過MQTT協議上傳至EMQX服務器，再由EMQX服務器將資料轉存至MySQL，這樣就可以讓后端人員很方便的去調取所需要的資料內容，

提示：以下是本篇文章正文內容，下面案例可供參考

一、專案準備

1、硬體準備

2、軟體準備

3、使用前須知

在專案開始前，要記得在云服務器的 “安全組/防火墻” 那邊開放寶塔的默認埠（8888），再順便把MQTT（1883）、WebSocket（8083）、EMQX（18083）、MySQL（3306）、phpMyAdmin（888）這些埠開放

二、寶塔面板安裝與配置

1、安裝寶塔面板

寶塔面板安裝鏈接：https://www.bt.cn/bbs/thread-19376-1-1.html
在云服務器上安裝寶塔面板，阿里云、騰訊云、華為云等云平臺均可，只需在下載的時候選對自己的云服務器所對應的系統及版本

我這里用的是阿里云的Ubuntu 18.04 LTS 云服務器

在云服務器中運行以下指令，自動安裝，安裝完后，在網頁端打開寶塔面板（xxx.xxx.xxx.xxx:8888）默認選擇LNMP
（注意，云服務器不能在安裝寶塔面板之前，安裝過（Apache、MySQL、PHP、Nginx）等環境，不然會導致出錯）

wget -O install.sh http://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh

安裝完寶塔后，即可通過外網面板地址進入登錄界面，使用“username”“password”登入寶塔面板
（若是覺得默認的 “username” 和 “password” 比較難記或者麻煩，也可以在寶塔面板的“面板設定”中進行修改）

2、創建資料庫

在寶塔面板中 “資料庫”，點擊“添加資料庫”新建一個資料庫
資料庫編碼格式中選擇 “utf8mb4”，避免出現編碼問題

3、新建表并設定欄位

3.1、Navicat For MySQL方法

• （如果沒有Navicat For MySQL，可跳過3.1小節，觀看3.2小節）

打開**“Navicat For MySQL”**來對資料庫進行管理操作，
首先建立一個新的連接

填寫好內容，不保證引數正確的話，可以測驗連接，確保測驗連接通過，提交后，即可成功創建連接，雙擊即可連接

“新建表”，在表中添加4個欄位

Ps：“ID” 欄位要勾上 “自動遞增” 方便排序，在資料表中看上去更直觀

并將表名命名為 “Temp_Humi_Data”

3.2、phpMyAdmin方法

點擊 “phpMyAdmin” 或 “管理” 進入phpMyAdmin進行資料庫管理

“新建表” 添加以下4個欄位，并將表命名為 “Temp_Humi_Data”

三、EMQ X Enterprise的安裝與規則創建

1、安裝EMQ X Enterprise

下載鏈接：https://www.emqx.cn/downloads
這里使用的是Ubuntu 18.04 LTS，我使用的是以下下載方法

若系統是Ubuntu 18.04LTS，選擇的引數與上圖一致，則可依次執行以下指令

• 下載Zip壓縮包

wget https://www.emqx.cn/downloads/enterprise/v4.2.5/emqx-ee-ubuntu18.04-4.2.5-x86_64.zip

• 解壓下載好的Zip壓縮包

unzip emqx-ee-ubuntu18.04-4.2.5-x86_64.zip

• 進入解壓后的檔案夾emqx

cd ./emqx

• 開啟EMQ X服務

./bin/emqx start

2、EMQ X Entterprise配置

在網址欄輸入“云服務器公網IP:18083”，進入 EMQ X Dashboard

在左側選中“規則引擎”—“資源”—“創建”，按照下圖填寫
（注意：帶有MySQL字樣的引數，都是在寶塔中新建資料庫的引數，要保證引數一致）

創建完資源后，接著創建規則，在 “SQL輸入” 中輸入以下SQL陳述句

SELECT
  payload.Temperature AS Temperature,
  payload.Humidity AS Humidity,
  payload.ts AS ts

FROM
  "Device_pubTopic/Temp_Humi" 

這里是獲取 “Device_pubTopic/Temp_Humi” payload中“Temperature”、“Humidity”、“ts”欄位的值

然后下去添加回應動作

其中 “使用資源” 為我們剛才創建的資源
SQL模板：

INSERT INTO 
	`Temp_Humi_Data` (`Temperature`, `Humidity`, `ts`)
VALUES 
	(${Temperature}, ${Humidity}, ${ts});

這里的意思是，向 “Temp_Humi_Data” 資料表中寫入“Temperature”、“Humidity”、“ts”的值

配置好后，本規則就可以將payload中的 “溫濕度及時間戳” 資料提取出來，并轉存到MySQL中相應資料庫的相應資料表中

3、MQTTX測驗

下載鏈接：https://mqttx.app/cn/
接下來使用MQTTX進行測驗，打開MQTTX，新建一個連接，填寫以下引數

在EMQ X Dashboard界面左側“工具”—“WebSocket”填寫 “”連接配置“” 的引數，并訂閱設備發布的主題 “Device_pubTopic/Temp_Humi”

注意事項：Client_ID不要和MQTTX中的Client_ID相同，不然會導致MQTTX連接后自動斷開

由下圖可看出訊息的發布與接收均正常
（紅色為MQTTX發布到服務器的JSON資料，藍色為服務器發布到MQTTX的JSON資料）

四、代碼

1、核心代碼

1.1、時間戳獲取

本專案獲取的時間戳是使用了蘇寧網站的授時網頁提供的
http://quan.suning.com/getSysTime.do

對此需要對網站內容進行JSON格式決議，提取出 “sysTime2” 欄位的值 “2021-04-20 20:29:03”

核心代碼

http.setTimeout(5000);	//設定5s超時
http.begin("http://quan.suning.com/getSysTime.do"); //HTTP初始化

int httpCode = http.GET();	//發送HTTP請求，回傳狀態碼
if (httpCode > 0) {
  Serial.printf("[HTTP] GET... code: %d\n", httpCode);
  if (httpCode == HTTP_CODE_OK) {
    //讀取回應內容
    JSON_DATA = http.getString();
    //這里回傳的是json格式的字串
    deserializeJson(Doc, JSON_DATA);
    JsonObject Obj = Doc.as<JsonObject>();
    Field_Val = Obj["sysTime2"];  //獲取Json資料中的欄位值Field(2020-04-20 20:29:03)
  }
} else {
  Serial.printf("[HTTP] GET... failed, error: %s\n", http.errorToString(httpCode).c_str());
}
http.end();	//接收完資料后，觸發end事件

1.2、設備訊息發布

利用
核心代碼

unsigned long now = millis();
if (now - lastMsg > 10000) {
  Get_Time();       //從網站獲得網頁內容
  Get_Temp_Humi();  //獲取溫濕度資料
  lastMsg = now;
  //String MQTT_Msg = "{\"Temperature\": \"%.1f\",\"Humidity\": \"%.1f\",\"ts\": \"%s\"}";
  //將訊息模板和溫濕度、時間戳資料合并到一起
  snprintf (Msg, Msg_LEN, MQTT_Msg, Temp, Humi, Field_Val.c_str());
  Serial.print("Publish message: ");
  Serial.println(Msg);
  client.publish(pubTopic, Msg);
}

1.3、資料庫訊息

2、完整代碼

#include <ESP8266WiFi.h>
#include <PubSubClient.h>
#include <ESP8266HTTPClient.h>
#include <ArduinoJson.h>
#include <dht.h>

/*連接的Wi-Fi、服務器地址*/
const char* ssid        = "SSID";
const char* password    = "PASSWORD";
const char* mqtt_server = "xxx.xxx.xxx.xxx";

/*設定發布和訂閱的主題*/
const char* subTopic = "Cloud_pubTopic";            //subTopic
const char* pubTopic = "Device_pubTopic/Temp_Humi"; //pubTopic

#define MQTT_Msg "{\"Temperature\": \"%.1f\",\"Humidity\": \"%.1f\",\"ts\": \"%s\"}"

WiFiClient espClient;
PubSubClient client(espClient);
long lastMsg = 0;
#define Msg_LEN 100
char Msg[Msg_LEN];

HTTPClient http;
String GetUrl = "http://quan.suning.com/getSysTime.do";
String JSON_DATA;
#define JSON_LEN 100
DynamicJsonDocument Doc(JSON_LEN);
String Target = "sysTime2";
String Field_Val = "";

dht DHT;
#define DHT11_PIN D2
float Temp, Humi;

void setup_wifi() {
  delay(10);
  // We start by connecting to a WiFi network
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  randomSeed(micros());

  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (int i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  Serial.println();

  // Switch on the LED if an 1 was received as first character
  if ((char)payload[0] == '1') {
    digitalWrite(BUILTIN_LED, LOW);   // Turn the LED on (Note that LOW is the voltage level
    // but actually the LED is on; this is because
    // it is active low on the ESP-01)
  } else {
    digitalWrite(BUILTIN_LED, HIGH);  // Turn the LED off by making the voltage HIGH
  }

}

void reconnect() {
  // Loop until we're reconnected
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Create a random client ID
    String clientId = "ESP8266Client-";
    clientId += String(random(0xffff), HEX);
    // Attempt to connect
    if (client.connect(clientId.c_str())) {
      Serial.println("connected");
      // Once connected, publish an announcement...
      //      client.publish(pubTopic, "hello world");
      // ... and resubscribe
      client.subscribe(subTopic);
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void setup() {
  pinMode(BUILTIN_LED, OUTPUT);     // Initialize the BUILTIN_LED pin as an output
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback);

  // 連接蘇寧網站的授時網頁
  http.setTimeout(5000);
  http.begin(GetUrl);
}

void loop() {

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

  unsigned long now = millis();
  if (now - lastMsg > 10000) {
    Get_Time();       //從網站獲得網頁內容
    Get_Temp_Humi();  //獲取溫濕度資料
    lastMsg = now;
    //String MQTT_Msg = "{\"Temperature\": \"%.1f\",\"Humidity\": \"%.1f\",\"ts\": \"%s\"}";    
    snprintf (Msg, Msg_LEN, MQTT_Msg, Temp, Humi, Field_Val.c_str());
    Serial.print("Publish message: ");
    Serial.println(Msg);
    client.publish(pubTopic, Msg);
  }
}

/**********
 * 讀取DHT11的溫濕度資料
 **********/
void Get_Temp_Humi() {
  int chk = DHT.read11(DHT11_PIN);
  switch (chk)
  {
    case DHTLIB_OK:
      Serial.print("OK,\t");
      break;
    case DHTLIB_ERROR_CHECKSUM:
      Serial.print("Checksum error,\t");
      break;
    case DHTLIB_ERROR_TIMEOUT:
      Serial.print("Time out error,\t");
      break;
    default:
      Serial.print("Unknown error,\t");
      break;
  }

  Temp = DHT.temperature;
  Humi = DHT.humidity;
}

void Get_Time() {
  int httpCode = http.GET();
  if (httpCode > 0) {
    Serial.printf("[HTTP] GET... code: %d\n", httpCode);
    if (httpCode == HTTP_CODE_OK) {
      //讀取回應內容
      JSON_DATA = http.getString();
      //這里回傳的是json格式的字串
      Field_Val = Get_JSON_Field(JSON_DATA, "sysTime2");  //獲取Json資料中的欄位值Field
    }
  } else {
    Serial.printf("[HTTP] GET... failed, error: %s\n", http.errorToString(httpCode).c_str());
  }
  http.end();
}

String Get_JSON_Field(String JSON_FORMAT_DATA, String Target) {
  // 將 JSON_FORMAT_DATA 替換成你想處理的字串
  deserializeJson(Doc, JSON_FORMAT_DATA);
  JsonObject Obj = Doc.as<JsonObject>();
  String Params = Obj[Target];  //將Target欄位的值提取出來賦給Params
  return Params;
}

五、總結

以上就是使用Wemos（ESP8266）通過MQTT將資料傳輸到服務器并轉存到MySQL中，本文僅僅簡單介紹了Wemos（ESP8266）上傳資料至云服務器的使用，如有寫的不好的地方，歡迎大家提點寶貴的建議，