本文共 15478 字，大约阅读时间需要 51 分钟。

关于如何在Windows下设置mqtt服务器请移步上一篇博文：

---

20210629更

python客户端使用pyqtgraph+pyqt5的方式进行绘图，将mqtt数据接收处理部分放到一个qthread上去，每次接收到数据后处理，然后通过信号的方式发送到主线程，主线程绑定一个槽函数，进行绘图的更新操作（代码见文末）

---

1.环境准备

1. mosqutto服务端程序，需要进行一些配置，重启服务，默认服务端口为1883
2. mqttx客户端程序，方便订阅和发布信息：
3. Arduino通过包管理器安装PubSubClient
4. esp32连接的网络和运行mosqutto服务程序的电脑处在同一个网段

2.arduino代码

/*********  @author: Wenqing Zhou (zhou.wenqing@qq.com)  @github: https://github.com/ouening    功能  ESP32搭建一个MQTT客户端，订阅主题"esp32/output"，以此控制led灯；发布主题  "esp32/dht11/temperature"和"esp32/dht11/humidity"将DHT11获取的温湿度数据推给  MQTT服务器，如果其他客户端在相同的MQTT服务器下订阅了该主题，便可获取对应的温度或者湿度  数据。    参考链接：Complete project details at https://randomnerdtutorials.com  *********/#include 
   
    #include 
    
     #include 
     
      #include 
      
       #include 
       
        #include 
        
         #include 
         
          // Replace the next variables with your SSID/Password combinationconst char* ssid = "wifi名称";const char* password = "wifi密码";// Add your MQTT Broker IP address, example:const char* mqtt_server = "192.168.28.87"; //先测试本机mqtt服务，该地址在windows下通过ipconfig查看，要和esp32连接在同一个网络//const char* mqtt_server = "YOUR_MQTT_BROKER_IP_ADDRESS";const char *id = "ESP32";const char *user = "kindy";const char *pass = "123456";WiFiClient espClient;PubSubClient client(espClient); // MQTT服务设置了非账号密码不能使用，所有在connect的时候要设置账号密码long lastMsg = 0;char msg[50];int value = 0;/* 设置DHT11 */#define DHTPIN 4 // Digital pin connected to the DHT sensor #define DHTTYPE DHT11 // DHT 11DHT_Unified dht(DHTPIN, DHTTYPE);float temperature = 0;float humidity = 0;// LED Pinconst int ledPin = 2;void setup() { Serial.begin(115200); pinMode(ledPin, OUTPUT); // default settings setup_wifi(); client.setServer(mqtt_server, 1883); client.setCallback(callback); // 绑定回调函数}void setup_wifi() { delay(10); // We start by connecting to a WiFi network Serial.println(); Serial.print("Connecting to "); Serial.println(ssid); WiFi.mode(WIFI_STA); // station mode WiFi.begin(ssid, password); while (WiFi.status() != WL_CONNECTED) { delay(1000); Serial.print("."); } Serial.println(""); Serial.println("WiFi connected"); Serial.println("IP address: "); Serial.println(WiFi.localIP()); /* ====== 初始化DHT11 ========*/ dht.begin(); Serial.println(F("DHTxx Unified Sensor Example")); // Print temperature sensor details. sensor_t sensor; dht.temperature().getSensor(&sensor); Serial.println(F("------------------------------------")); Serial.println(F("Temperature Sensor")); Serial.print (F("Sensor Type: ")); Serial.println(sensor.name); Serial.print (F("Driver Ver: ")); Serial.println(sensor.version); Serial.print (F("Unique ID: ")); Serial.println(sensor.sensor_id); Serial.print (F("Max Value: ")); Serial.print(sensor.max_value); Serial.println(F("°C")); Serial.print (F("Min Value: ")); Serial.print(sensor.min_value); Serial.println(F("°C")); Serial.print (F("Resolution: ")); Serial.print(sensor.resolution); Serial.println(F("°C")); Serial.println(F("------------------------------------")); // Print humidity sensor details. dht.humidity().getSensor(&sensor); Serial.println(F("Humidity Sensor")); Serial.print (F("Sensor Type: ")); Serial.println(sensor.name); Serial.print (F("Driver Ver: ")); Serial.println(sensor.version); Serial.print (F("Unique ID: ")); Serial.println(sensor.sensor_id); Serial.print (F("Max Value: ")); Serial.print(sensor.max_value); Serial.println(F("%")); Serial.print (F("Min Value: ")); Serial.print(sensor.min_value); Serial.println(F("%")); Serial.print (F("Resolution: ")); Serial.print(sensor.resolution); Serial.println(F("%")); Serial.println(F("------------------------------------"));}void callback(char* topic, byte* message, unsigned int length) { Serial.print("Message arrived on topic: "); Serial.print(topic); Serial.print(". Message: "); String messageTemp; for (int i = 0; i < length; i++) { Serial.print((char)message[i]); messageTemp += (char)message[i]; } Serial.println(); // Feel free to add more if statements to control more GPIOs with MQTT // If a message is received on the topic esp32/output, you check if the message is either "on" or "off". // Changes the output state according to the message if (String(topic) == "esp32/output") { Serial.print("Changing output to "); if(messageTemp == "on"){ Serial.println("on"); digitalWrite(ledPin, HIGH); } else if(messageTemp == "off"){ Serial.println("off"); digitalWrite(ledPin, LOW); } }}/* 重连mqtt服务器 */void reconnect() { // Loop until we're reconnected while (!client.connected()) { Serial.print("Attempting MQTT connection..."); //MQTT服务设置了非账号密码不能使用，所有在connect的时候要设置账号密码 if (client.connect(id, user, pass)) { Serial.println("connected"); // 订阅mqtt主题 client.subscribe("esp32/output"); } 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(); long now = millis(); if (now - lastMsg > 5000) { lastMsg = now; char tempString[8]; // Convert the value to a char array char humString[8]; // 获取温度数据 sensors_event_t event; dht.temperature().getEvent(&event); if (isnan(event.temperature)) { Serial.println(F("Error reading temperature!")); } else { dtostrf((float)event.temperature,2,2,tempString); // convert to String Serial.print("Temperature: "); Serial.println(tempString); client.publish("esp32/dht11/temperature", tempString); // 发布信息，第一个参数是主题 } // Get humidity event and print its value. dht.humidity().getEvent(&event); if (isnan(event.relative_humidity)) { Serial.println(F("Error reading humidity!")); } else { dtostrf((float)event.relative_humidity,2,2,humString); Serial.print("Humidity: "); Serial.println(humString); client.publish("esp32/dht11/humidity", humString); } }}
         
        
       
      
     
    
   

3.python mqtt客户端数据可视化

借助包paho-mqtt，可以在用python创建mqtt客户端，订阅相关主题，并用matplotlib可视化，代码如下：

# -*- coding: utf-8 -*-"""Created on Sun Jun 20 11:28:12 2021@author: Wenqing Zhou (zhou.wenqing@qq.com)@github: https://github.com/ouening"""# 参考链接：# （1）https://www.cnblogs.com/chenpingzhao/p/11383856.html# （2）https://www.emqx.cn/blog/how-to-use-mqtt-in-python# 除了以下通过python程序接收mqtt订阅信息外，还可以下载相关的mqtt客户端程序，例如https://mqttx.app/zhimport randomfrom paho.mqtt import client as mqtt_clientimport matplotlib.pyplot as pltimport datetimeplt.rcParams.update({
       "font.family": "serif",    "font.serif": ["Times New Roman"], #Times New Roman    "font.size":11,    "legend.fancybox": False,    "legend.frameon": True,    "legend.framealpha": 1.0,    "legend.shadow": False,    "legend.facecolor": 'white',    "legend.edgecolor": 'none',  # 'none', 0.8    "legend.fontsize": "small",  # {'xx-small', 'x-small', 'small', 'medium', 'large', 'x-large', 'xx-large'}    "text.usetex": False,    })# broker = 'broker.emqx.io'# broker = "192.168.28.87"broker = '公网服务器ip地址'port = 1883username = 'kindy'passwd = 'kindy'# topic = "mytopic/test" # esp32程序设定的推送主题topic1 = "esp32/dht11/humidity"topic2 = "esp32/dht11/temperature"# generate client ID with pub prefix randomlyclient_id = f'esp32'count = 0x1, x2 = [], []# 存储数据sub_dict = {
   topic1:[], topic2:[]}def connect_mqtt() -> mqtt_client:    def on_connect(client, userdata, flags, rc):        if rc == 0:            print("Connected to MQTT Broker!")        else:            print("Failed to connect, return code %d\n", rc)    client = mqtt_client.Client(client_id)    client.on_connect = on_connect    # 连接mqtt服务器    client.username_pw_set(username, password=passwd)    client.connect(broker, port)    return clientdef on_message(client, userdata, msg):    # 存储数据    global count    print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")    # 绘图    plt.ion()    count = count + 1    if msg.topic==topic1:        sub_dict[msg.topic].append(float(msg.payload.decode()))        x1.append(datetime.datetime.now()) # add timestamp    if msg.topic==topic2:        sub_dict[msg.topic].append(float(msg.payload.decode()))        x2.append(datetime.datetime.now()) # add timestamp    # 只存储100个数据    if (len(x1)==100) and (len(sub_dict[topic1])==100):        x1.pop(0)        sub_dict[topic1].pop(0)    if (len(x2)==100) and (len(sub_dict[topic2])==100):        x2.pop(0)        sub_dict[topic2].pop(0)    # y_hum.append(sub_dict[topic1]) # add temperature    # print('Temp,Vbat,A0,count',d,count)    plt.clf()    plt.plot(x1, sub_dict[topic1], label=topic1)    plt.plot(x2, sub_dict[topic2], label=topic2)    plt.gcf().autofmt_xdate()    plt.xlabel('time')    plt.ylabel('Temperature')    plt.legend(loc='best')    plt.grid('on')    plt.show()    plt.pause(0.01)def subscribe(client: mqtt_client):    client.subscribe(topic1)    client.subscribe(topic2)    client.on_message = on_messagedef run():    client = connect_mqtt()    subscribe(client)    client.loop_forever()if __name__ == '__main__':    run()

效果如下：

4.MQTTX客户端

MQTTX设置连接：

MQTTX订阅主题：

MQTTX发布信息：

本例子是用在本地电脑测试的 ，如果运行mosqutto服务程序的是在远端的服务器，或者其他厂商提供的物联网服务器，相关设置应该都大同小异，主要是***ip地址、端口，账号密码***等；另外，从mqtt可选的发送和接收数据类型来看，除了普通的文本数据（plaintext）外，还可以发送json等高级数据格式。

已知问题：python构建mqtt客户端再用matplotlib可视化的程序会一直阻塞，对图形不好控制，下一步应该考虑多进程，将绘图功能放到其他子线程，不要干扰主线程；另外可以采用pyqt5或者pyside进行界面封装。

# -*- coding: utf-8 -*-"""Created on Tue Jun 29 14:27:51 2021@author: Wenqing Zhou (zhou.wenqing@qq.com)@github: https://github.com/ouening"""import sysimport ctypesimport timeimport pyqtgraph as pgimport threadingimport serialfrom collections import dequefrom PyQt5.QtCore import *from PyQt5.QtGui import *from PyQt5.QtWidgets import *import datetimefrom matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvasimport matplotlib.pyplot as pltfrom pyqtgraph import DateAxisItemfrom paho.mqtt import client as mqtt_client__version__ = '1.0'broker = 'xx.xx.xx.xx'port = 1883username = 'kindy'passwd = 'xxxx'# topic = "mytopic/test" # esp32程序设定的推送主题topic1 = "esp32/dht11/temperature"topic2 = "esp32/dht11/humidity"# generate client ID with pub prefix randomlyclient_id = f'esp32'class MQTTWorker(QThread):    data_signal = pyqtSignal(object)    def __init__(self):        super().__init__()        # 初始化mqtt        self.count = 0        self.history = 7200        self.sub_dict = {
   'x1':[], topic1:[], 'x2':[], topic2:[]}        def on_connect(client, userdata, flags, rc):            if rc == 0:                print("Connected to MQTT Broker!")            else:                print("Failed to connect, return code %d\n", rc)            self.client = mqtt_client.Client(client_id)        self.client.on_connect = on_connect            # 连接mqtt服务器        self.client.username_pw_set(username, password=passwd)        self.client.connect(broker, port)        # 订阅主题        self.client.subscribe(topic1)        self.client.subscribe(topic2)        # 信息接收回调函数        def on_message(client, userdata, msg):            # 存储数据            print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")            # 绘图            self.count += 1                    if msg.topic==topic1:                self.sub_dict[msg.topic].append(float(msg.payload.decode()))                self.sub_dict['x1'].append(time.time()) # add timestamp            if msg.topic==topic2:                self.sub_dict[msg.topic].append(float(msg.payload.decode()))                self.sub_dict['x2'].append(time.time()) # add timestamp            # 只存储100个数据            if (len(self.sub_dict['x1'])==self.history) and (len(self.sub_dict[topic1])==self.history):                self.sub_dict['x1'].pop(0)                self.sub_dict[topic1].pop(0)            if (len(self.sub_dict['x2'])==self.history) and (len(self.sub_dict[topic2])==self.history):                self.sub_dict['x2'].pop(0)                self.sub_dict[topic2].pop(0)                self.data_signal.emit(self.sub_dict) # 发送信号        self.client.on_message = on_message    # 停止qthread线程    def stop(self):        self.terminate()        print("QThread terminated")    # mqtt客户端循环接收数据    def run(self):        self.client.loop_forever()class MainWindow(QMainWindow):    newdata = pyqtSignal(object) # 创建信号    def __init__(self, filename=None, parent=None):        super(MainWindow, self).__init__(parent)        self.setWindowTitle('温湿度数据采集')        self.setWindowIcon(QIcon(r"D:\Github\bouncescope\smile.ico"))        self.connected = False        # 实例化线程        self.mqtt_worker = MQTTWorker()        # QTimer.singleShot(0, self.startThread)        self.btn = QPushButton('点击运行！')        self.label = QLabel("实时获取温湿度数据")        self.label.setAlignment(Qt.AlignCenter)        self.data_label = QLabel("Data") # label显示数据        self.pw = pg.PlotWidget()        self.pw_hum = pg.PlotWidget()        # setup pyqtgraph        self.init_pg_temp()  # 温度        self.init_pg_hum()  # 湿度                # 添加布局，注意布局的嵌套        vb = QVBoxLayout()        hb = QHBoxLayout()        vb.addWidget(self.label)        vb.addWidget(self.btn)        hb.addWidget(self.pw)        hb.addWidget(self.pw_hum)        vb.addLayout(hb)        vb.addWidget(self.data_label)        # 添加一个widget，因为widget才有setLayout方法，MainWindow没有        self.cwidget = QWidget()        self.cwidget.setLayout(vb)        self.setCentralWidget(self.cwidget)        self.mqtt_worker.data_signal.connect(self.process_mqtt) # 连接处理，mqtt信号的槽函数        self.btn.clicked.connect(self.startThread) # 启动mqtt线程    def init_pg_temp(self):        # 设置图表标题        self.pw.setTitle("温度变化趋势",                         color='008080',                         size='12pt')        # 设置上下左右的label        self.pw.setLabel("left","气温(摄氏度)")        self.pw.setLabel("bottom","时间")        # 设置Y轴 刻度 范围        # self.pw.setYRange(min=10, max=50)        # 显示表格线        self.pw.showGrid(x=True, y=True)        # 背景色改为白色        self.pw.setBackground('w')        # 居中显示 PlotWidget        # self.setCentralWidget(self.pw)        axis = DateAxisItem()        self.pw.setAxisItems({
   'bottom':axis})        self.curve_temp = self.pw.getPlotItem().plot(            pen=pg.mkPen('r', width=1)        )    def init_pg_hum(self):                # 设置图表标题        self.pw_hum.setTitle("湿度度变化趋势",                         color='008080',                         size='12pt')        # 设置上下左右的label        self.pw_hum.setLabel("left","湿度")        self.pw_hum.setLabel("bottom","时间")        # 设置Y轴 刻度 范围        # self.pw_hum.setYRange(min=10, # 最小值        #                   max=100)  # 最大值        # 显示表格线        self.pw_hum.showGrid(x=True, y=True)        # 背景色改为白色        self.pw_hum.setBackground('w')        axis = DateAxisItem()        self.pw_hum.setAxisItems({
   'bottom':axis})        self.curve_hum = self.pw_hum.getPlotItem().plot(            pen=pg.mkPen('b', width=1)        )    # 处理mqtt数据 sub_dict = {'x1':[], topic1:[], 'x2':[], topic2:[]}    def process_mqtt(self, sub_dict):        try:            self.data_label.setText(str(datetime.datetime.now()))            self.curve_temp.setData(sub_dict['x1'], sub_dict[topic1])            self.curve_hum.setData(sub_dict['x2'], sub_dict[topic2])        except:            print(sub_dict)    def startThread(self):        print('Start lisnening to the mqtt')        self.btn.setEnabled(False)        self.mqtt_worker.start()            def stopThread(self):        self.mqtt_worker.stop()        print ('Stop the thread...')    def closeEvent(self, event):        if self.okToContinue():             event.accept()             self.stopThread()        else:            event.ignore()    def okToContinue(self):        return Trueif __name__ == '__main__':    app = QApplication(sys.argv)    win = MainWindow()    win.show()    app.exec_()

已知问题：mqtt服务器连接断开后，程序就无法继续获取数据绘图了，有待解决