ESP8266通過(guò)mqtt保存數(shù)據(jù)到mysql
0.硬件和軟件
ESP8266、DHT11、EMQX（企業(yè)版）、MYSQL（5.6）、服務(wù)器ubuntu2018
1.EMQX安裝
下載地址
https://www.emqx.com/zh/downloads/enterprise選擇一個(gè)版本
選擇版本復(fù)制下載鏈接
1.下載安裝包
wget https://www.emqx.com/zh/downloads/enterprise/v4.4.17/emqx-ee-4.4.17-otp24.3.4.2-1-ubuntu20.04-amd64.deb2.執(zhí)行安裝
sudo apt install ./emqx-ee-4.4.17-otp24.3.4.2-1-ubuntu20.04-amd64.deb3.啟動(dòng)服務(wù)
emqx start按步驟復(fù)制上面三個(gè)命令執(zhí)行
安裝成功后，開(kāi)放以下端口：8883、1883、8083、8084、18083
然后訪問(wèn)EMQX后臺(tái)：服務(wù)器IP:18083
由于企業(yè)版收費(fèi)使用試用的license連接只能為10個(gè)
（1）安裝MQTT庫(kù)
打開(kāi)arduino點(diǎn)擊庫(kù)管理，然后搜索pubsubclient
安裝PubSubClient
（2）安裝DHT11溫濕度傳感器庫(kù)
搜索simpleDHT
安裝
（3）安裝WIFImanager庫(kù)，這個(gè)庫(kù)可以通過(guò)Web端配網(wǎng)
搜索WIFImanager
安裝ESP32Wifimanager
下面代碼每隔1秒通過(guò)MQTT上傳溫濕度數(shù)據(jù)，每隔200ms獲取訂閱的消息（用于遠(yuǎn)程控制）通過(guò)ticker庫(kù)實(shí)現(xiàn)多任務(wù)處理，讓上傳數(shù)據(jù)和獲取訂閱同時(shí)執(zhí)行，具體代碼：
#include <ESP8266WiFi.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>
#include <SimpleDHT.h>
#include <DNSServer.h>
#include <ESP8266WebServer.h>
#include <WiFiManager.h>
#include <Ticker.h>
// MQTT 信息
const char* mqttServer = "mqtt.xxxxx.com";//MQTT服務(wù)器地址
const int mqttPort = 1883; // 默認(rèn) MQTT 端口
const char* mqttPublishTopic = "wen"; // MQTT 發(fā)布主題
const char* mqttSubscribeTopic = "top"; // MQTT 訂閱主題
// 溫濕度傳感器引腳
const int dhtPin = 2; // 假設(shè)溫濕度傳感器連接到 D4
WiFiClient espClient;
// 創(chuàng)建 MQTT 客戶端
PubSubClient client(espClient);
// 創(chuàng)建 DHT 溫濕度傳感器對(duì)象
SimpleDHT11 dht11(dhtPin);
// 定義 Ticker 對(duì)象
Ticker publishDataTicker;
Ticker subscribeDataTicker;
// 定義定時(shí)器回調(diào)函數(shù)
void publishDataTask();
void subscribeDataTask();
void setup() {
?// 初始化串口
?Serial.begin(115200);
?// Web配網(wǎng)
?// 建立WiFiManager對(duì)象
?WiFiManager wifiManager;
?wifiManager.resetSettings();//清除上一次保存的WIFI賬號(hào)和密碼
?wifiManager.autoConnect("Web配網(wǎng)的WIFI名稱");
?// 連接到 MQTT 服務(wù)器
?client.setServer(mqttServer, mqttPort);
?client.setCallback(callback);
?while (!client.connected()) {
? ?Serial.println("連接MQTT服務(wù)器中...");
? ?if (client.connect("MQTT連接名稱", mqttUser, mqttPassword)) {
? ? ?Serial.println("連接MQTT成功");
? ?} else {
? ? ?Serial.print("連接失?。?");
? ? ?Serial.print(client.state());
? ? ?delay(2000);
? ?}
?}
?// 訂閱 MQTT 主題
?client.subscribe(mqttSubscribeTopic);
?// 啟動(dòng)定時(shí)器
?publishDataTicker.attach_ms(1000, publishDataTask); // 每隔 1 秒執(zhí)行一次 publishDataTask 函數(shù)
?subscribeDataTicker.attach_ms(100, subscribeDataTask); // 每隔 100 毫秒執(zhí)行一次 subscribeDataTask 函數(shù)
}
void loop() {
?// 循環(huán)函數(shù)為空，所有任務(wù)通過(guò)定時(shí)器進(jìn)行處理
}
// 定時(shí)器回調(diào)函數(shù)，發(fā)布溫濕度數(shù)據(jù)到 MQTT
void publishDataTask() {
?// 讀取溫濕度數(shù)據(jù)
?byte temperature = 0;
?byte humidity = 0;
?int err = SimpleDHTErrSuccess;
?if ((err = dht11.read(&temperature, &humidity, NULL)) != SimpleDHTErrSuccess) {
? ?Serial.print("溫濕度傳感器異常:"); Serial.print(SimpleDHTErrCode(err));
? ?Serial.print(","); Serial.println(SimpleDHTErrDuration(err)); delay(1000);
? ?return;
?}
?// 將溫濕度數(shù)據(jù)轉(zhuǎn)換為json字符串
? ?DynamicJsonDocument data(256);
? ?data["wen"] = String(temperature);
? ?data["shi"] = String(humidity);
? ?char json_string[256];
? ?serializeJson(data, sendjson);
?// 發(fā)布溫濕度數(shù)據(jù)到 MQTT
?client.publish(mqttPublishTopic, sendjson, false);
}
// 定時(shí)器回調(diào)函數(shù)，檢查 MQTT 訂閱的消息
void subscribeDataTask() {
?// 檢查 MQTT 訂閱的消息
?client.loop();
}
// MQTT 消息回調(diào)函數(shù)
void callback(char* topic, byte* payload, unsigned int length) {
?// 處理 MQTT 消息
?// 將接收到的消息轉(zhuǎn)換為字符串
?String message;
?for (int i = 0; i < length; i++) {
? ?message += (char)payload[i];
?}
?// 輸出接收到的消息
?Serial.print("收到消息: ");
?Serial.println(message);
}燒錄代碼，復(fù)位后，手機(jī)連接esp配網(wǎng)wifi按照要求數(shù)據(jù)密碼賬號(hào)后等待esp連接到emqx
測(cè)試連接，訪問(wèn)mqtt消息調(diào)試工具M(jìn)QTT X Web: 在線的 MQTT Websocket 客戶端工具
https://mqttx.app/zh/web
點(diǎn)擊新建連接，輸入服務(wù)器地址點(diǎn)擊連接
添加訂閱接收數(shù)據(jù)
測(cè)試接收到的數(shù)據(jù)
3.保存到mysql數(shù)據(jù)庫(kù)
添加數(shù)據(jù)庫(kù)連接
點(diǎn)擊創(chuàng)建，選擇mysql
然后輸入賬號(hào)密碼，點(diǎn)擊確定
根據(jù)上面?zhèn)魃蟻?lái)的json數(shù)據(jù)添加sql模板
SELECT
timestamp as time_up,payload.wen as wen, payload.shi as shi ?
FROM ?
"wen" ?
添加響應(yīng)動(dòng)作，選擇剛剛設(shè)置的sql數(shù)據(jù)庫(kù)
添加模板
insert into tmp(time_up, wen, shi) values (FROM_UNIXTIME(${time_up}/1000), ${wen}, ${shi}) 最后測(cè)試當(dāng)客戶端發(fā)送數(shù)據(jù)上來(lái)，數(shù)據(jù)被保存在sql數(shù)據(jù)庫(kù)中
