1.EMQ上连接服务器，主机地址填ESP8266上传数据的服务器公网地址，等会在ESP8266的代码段上可以看见相应的主机地址。

 

 2.订阅ESP8266上的相同的topic（ESP作为Sub，EMQ作为Pub）

 

 

 3.上传代码给ESP8266，上传完成即实现向阿里云传输数据，下面步骤是利用python将数据写入txt文件并保存到本地

ssid为WiFi名

password 为WiFi密码

#include <ESP8266WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>

//esp8266引脚

#define D0  16
#define D1  5
#define D2  4
#define D3  0
#define D4  2
#define D5  14
#define D6  12
#define D7  13
#define D8  15
#define D9  3
#define D10  1


const char* ssid = "Xiaomi_C6F4";
const char* password = "10241024";
//填入公网地址如192.168.0.0这样的不用加http这些前缀
const char* mqtt_server = ""; // 使用HIVEMQ 的信息中转服务


const char* TOPIC = "/test/msg/";                     // 订阅信息主题
const char* client_id = "Mymqtt";                   // 标识当前设备的客户端编号，随便填

float RH,T;
DHT dht(D3,DHT11, 15); //温湿度引脚
WiFiClient espClient;                                                         // 定义wifiClient实例
PubSubClient client(espClient);                                         // 定义PubSubClient的实例
long lastMsg = 0;                                                               // 记录上一次发送信息的时长

void setup() {

  dht.begin();
  Serial.begin(115200); 
  setup_wifi();                                                                    //执行Wifi初始化，下文有具体描述
  client.setServer(mqtt_server, 1883);                              //设定MQTT服务器与使用的端口，1883是默认的MQTT端口
  client.setCallback(callback);                                          //设定回调方式，当ESP8266收到订阅消息时会调用此方法
}

void setup_wifi() {

  delay(10);
  // 板子通电后要启动，稍微等待一下让板子点亮
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

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

  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();
  
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Attempt to connect
    if (client.connect(client_id)) {
      Serial.println("connected");
      // 连接成功时订阅主题
      client.subscribe(TOPIC);
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void loop() {

  if (!client.connected()) {
    reconnect();
  }
  client.loop();
   RH = dht.readHumidity();   //读取湿度数据                       
   T = dht.readTemperature();       //读取温度数据
//   RH_sum+=RH;
//   T_sum+=T;                    
  long now = millis();
  if (now - lastMsg > 3000) {
    lastMsg = now;
    String msg = "Tempature: " + (String)T + "Humidity: " + (String)RH;
    client.publish("/test/msg/", msg.c_str());//"'temperature':26,'Humidity':30"
  }
}

连线图

 

 4.利用python程序读取

import paho.mqtt.client as mqtt
import time 
MQTTHOST = ""
MQTTPORT = 1883
mqttClient = mqtt.Client()
 
 
# 连接MQTT服务器
def on_mqtt_connect():
    mqttClient.connect(MQTTHOST, MQTTPORT, 60)
    mqttClient.loop_start()
 
 
# publish 消息
def on_publish(topic, payload, qos):
    mqttClient.publish(topic, payload, qos)
 
# 消息处理函数
def on_message_come(lient, userdata, msg):
    
    string = msg.topic + " " + ":" + str(msg.payload)
    data = open(r"test.txt","w")
    data.write(string)
    data.close()
    time.sleep(5)
    print(string)

 
 
# subscribe 消息
def on_subscribe():
    mqttClient.subscribe("/test/msg/", 0)
    mqttClient.on_message = on_message_come # 消息到来处理函数
 
 
def main():
    on_mqtt_connect()
    on_publish("/test/server", "Hello Python!", 1)
    on_subscribe()
    while True:
        pass
 
 
 
if __name__ == '__main__':
    main()

 

效果图

 

text.txt文件里面的数据一直在变化，可以进行数据的读取以及其他操作