ESP32+MQTT+MySQL实现发布订阅【气味数据收集】

int mq_pin = 33;
void setup() {
 // put your setup code here, to run once:
 Serial.begin(9600);   //9600 bps
 //Serial.println("CLEARDATA");
 //Serial.println("LABEL,VALUE");
}
void loop() {
 // put your main code here, to run repeatedly:
 int val;
 val=analogRead(33);
 //Serial.println(val ,DEC);
 //Serial.print("DATA, TIMER,");
 Serial.println(val);
 delay(500);
}

#include <WiFi.h>
#include <WiFiClient.h>
#include <PubSubClient.h>
#include <Ticker.h>    
// 设置wifi接入信息(请根据您的WiFi信息进行修改)
const char* ssid = "xxxxxx";     //WIFI名称
const char* password = "xxxxxxx";   //WIFI密码
const char* mqttServer = "xxxxxxxx";    //MQTT服务器IP
Ticker ticker;   //定时执行任务
WiFiClient wifiClient;
PubSubClient mqttClient(wifiClient);
int count;    // Ticker计数用变量
int mq_pin = 33;   //引脚号
void setup() {
 Serial.begin(9600);
 //设置ESP32工作模式为无线终端模式
 WiFi.mode(WIFI_STA);
 // 连接WiFi
 connectWifi();
 // 设置MQTT服务器和端口号
 mqttClient.setServer(mqttServer, 1883);
 // 连接MQTT服务器
 connectMQTTServer();
 // Ticker定时对象
 ticker.attach(1, tickerCount); 
}
void loop() { 
 // delay(500);
 if (mqttClient.connected()) { // 如果开发板成功连接服务器
   // 每隔1秒钟发布一次信息
   if (count >= 2){
     pubMQTTmsg();
     count = 0;
   }    
   // 保持心跳
   mqttClient.loop();
 } else {                  // 如果开发板未能成功连接服务器
   connectMQTTServer();    // 则尝试连接服务器
 }
}
void tickerCount(){
 count++;
}
void connectMQTTServer(){
 String clientId = "slldxdb";
 // 连接MQTT服务器
 if (mqttClient.connect(clientId.c_str())) { 
   Serial.println("MQTT Server Connected.");
   Serial.println("Server Address: ");
   Serial.println(mqttServer);
   Serial.println("ClientId:");
   Serial.println(clientId);
 } else {
   Serial.print("MQTT Server Connect Failed. Client State:");
   Serial.println(mqttClient.state());   //打印连接返回码
   delay(3000);
 }   
}
// 发布信息
void pubMQTTmsg(){
 static int value; // 客户端发布信息用数字
 String topicString = "mqtt-mark";   //建立发布主题
 char publishTopic[topicString.length() + 1];  
 strcpy(publishTopic, topicString.c_str());
 int val;
 val=analogRead(mq_pin);
 // 建立发布信息
 String messageString = String(val);
 char publishMsg[messageString.length() + 1];   
 strcpy(publishMsg, messageString.c_str());
 // 实现ESP32向主题发布信息
 if(mqttClient.publish(publishTopic, publishMsg)){
   Serial.println("Publish Topic:");Serial.println(publishTopic);
   Serial.println("Publish message:");Serial.println(publishMsg);    
 } else {
   Serial.println("Message Publish Failed."); 
 }
}
// ESP32连接wifi
void connectWifi(){
 WiFi.begin(ssid, password);
 //等待WiFi连接,成功连接后输出成功信息
 while (WiFi.status() != WL_CONNECTED) {
   delay(500);
   Serial.print(".");
 }
 Serial.println("");
 Serial.println("WiFi Connected!");  
 Serial.println(""); 
}

import json
import paho.mqtt.client as mqtt
import pymysql
"""rc 值 
0 连接成功
1 协议版本错误
2 无效的客户端标识
3 服务器无法使用
4 错误的用户名或密码
5 未经授权
"""
#打开数据库连接 【自行填写】
mysql_conn = pymysql.connect(user="xxx",password="xxx",host="xxx",database="xxx",port=3306)
def on_connect(client, userdata, flags, rc):
   return rc
def on_message(client, userdata, publishMsg):
   # data = publishMsg.payload
   # print(json.loads(data))
   msg = publishMsg.payload.decode('gb2312')
   print(msg)
   ##存数据到数据库中
   cursor = mysql_conn.cursor()
   insert_sql = 'insert into test (value) VALUES ("%s")' %(msg)
   cursor.execute(insert_sql)
   # 提交到数据库执行
   mysql_conn.commit()
   cursor.close()  # 先关闭游标
def main():
   client = mqtt.Client()
   client.username_pw_set(username='', password='')
   client.on_connect = on_connect
   client.on_message = on_message
   rc = client.connect(host='xxxx', port=1883, keepalive=15)    ##自行填写服务器地址信息
   print('rc', rc)
   if rc == 0:
       client.subscribe('mqtt-mark')    ##订阅主题
   else:
       print('连接失败')
       raise Exception
   # 无限循环收发消息
   client.loop_forever()
if __name__ == '__main__':
   main()