将 paho-mqtt 与 PyQt 一起使用

Question

如何使用 QThread 的回调函数？

回调函数 on_message 不打印任何数据。 在 run() 中，我连接到 mqtt-broker 并订阅主题。当我收到新消息时，on_message 必须有效。

简单的 QT 应用示例。与简单 QLCD 连接的更改值事件。订阅主题取自dashboard

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import sys
from PyQt5.QtCore import Qt, QThread, pyqtSignal
from PyQt5.QtWidgets import (QWidget, QLCDNumber, QSlider,
                             QVBoxLayout, QApplication)
from random import uniform, normalvariate, randint
import time
import paho.mqtt.client as mqtt
import paho.mqtt.subscribe as subscribe

class Example(QWidget):

    def __init__(self):
        super().__init__()

        self.initUI()

    def initUI(self):

        lcd = QLCDNumber(self)
        self.data=DataThread()
        vbox = QVBoxLayout()
        vbox.addWidget(lcd)

        self.setLayout(vbox)
        self.data.valueChanged.connect(lcd.display)
        self.setGeometry(300, 300, 250, 150)
        self.setWindowTitle('Callback')
        self.data.start()
        self.show()

class DataThread(QThread):
    valueChanged = pyqtSignal(object)
    val=0

    def on_message(mqttc, obj, msg):
        mstr=msg.payload.decode("ascii")
        val=mstr
        self.valueChanged.emit(val)

    def on_subscribe(mqttc, obj, mid, granted_qos):
        print("Subscribed: "+str(mid)+" "+str(granted_qos))

    def run(self):
        msg = subscribe.simple("shok2", hostname="broker.hivemq.com")
        print("%s %s" % (msg.topic, msg.payload))
        mqttc = mqtt.Client()
        mqttc.on_subscribe = self.on_subscribe
        mqttc.on_message = self.on_message
        mqttc.connect("broker.hivemq.com")
        print("subsrcibe")
        mqttc.subscribe("shok2")
        mqttc.loop_forever()

if __name__ == '__main__':

    app = QApplication(sys.argv)
    ex = Example()
    sys.exit(app.exec_())

Answer 1

没有必要使用线程，只要遵循 Qt 的逻辑：让我们使用信号。在下面的代码中，我展示了我正在实现的 MQTT 客户端库的改进，它至少到目前为止实现了你想要的，在接下来的几天里，我将添加更多功能。

from PyQt5 import QtCore, QtWidgets
import paho.mqtt.client as mqtt


class MqttClient(QtCore.QObject):
    Disconnected = 0
    Connecting = 1
    Connected = 2

    MQTT_3_1 = mqtt.MQTTv31
    MQTT_3_1_1 = mqtt.MQTTv311

    connected = QtCore.pyqtSignal()
    disconnected = QtCore.pyqtSignal()

    stateChanged = QtCore.pyqtSignal(int)
    hostnameChanged = QtCore.pyqtSignal(str)
    portChanged = QtCore.pyqtSignal(int)
    keepAliveChanged = QtCore.pyqtSignal(int)
    cleanSessionChanged = QtCore.pyqtSignal(bool)
    protocolVersionChanged = QtCore.pyqtSignal(int)

    messageSignal = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(MqttClient, self).__init__(parent)

        self.m_hostname = ""
        self.m_port = 1883
        self.m_keepAlive = 60
        self.m_cleanSession = True
        self.m_protocolVersion = MqttClient.MQTT_3_1

        self.m_state = MqttClient.Disconnected

        self.m_client =  mqtt.Client(clean_session=self.m_cleanSession,
            protocol=self.protocolVersion)

        self.m_client.on_connect = self.on_connect
        self.m_client.on_message = self.on_message
        self.m_client.on_disconnect = self.on_disconnect


    @QtCore.pyqtProperty(int, notify=stateChanged)
    def state(self):
        return self.m_state

    @state.setter
    def state(self, state):
        if self.m_state == state: return
        self.m_state = state
        self.stateChanged.emit(state) 

    @QtCore.pyqtProperty(str, notify=hostnameChanged)
    def hostname(self):
        return self.m_hostname

    @hostname.setter
    def hostname(self, hostname):
        if self.m_hostname == hostname: return
        self.m_hostname = hostname
        self.hostnameChanged.emit(hostname)

    @QtCore.pyqtProperty(int, notify=portChanged)
    def port(self):
        return self.m_port

    @port.setter
    def port(self, port):
        if self.m_port == port: return
        self.m_port = port
        self.portChanged.emit(port)

    @QtCore.pyqtProperty(int, notify=keepAliveChanged)
    def keepAlive(self):
        return self.m_keepAlive

    @keepAlive.setter
    def keepAlive(self, keepAlive):
        if self.m_keepAlive == keepAlive: return
        self.m_keepAlive = keepAlive
        self.keepAliveChanged.emit(keepAlive)

    @QtCore.pyqtProperty(bool, notify=cleanSessionChanged)
    def cleanSession(self):
        return self.m_cleanSession

    @cleanSession.setter
    def cleanSession(self, cleanSession):
        if self.m_cleanSession == cleanSession: return
        self.m_cleanSession = cleanSession
        self.cleanSessionChanged.emit(cleanSession)

    @QtCore.pyqtProperty(int, notify=protocolVersionChanged)
    def protocolVersion(self):
        return self.m_protocolVersion

    @protocolVersion.setter
    def protocolVersion(self, protocolVersion):
        if self.m_protocolVersion == protocolVersion: return
        if protocolVersion in (MqttClient.MQTT_3_1, MQTT_3_1_1):
            self.m_protocolVersion = protocolVersion
            self.protocolVersionChanged.emit(protocolVersion)

    #################################################################
    @QtCore.pyqtSlot()
    def connectToHost(self):
        if self.m_hostname:
            self.m_client.connect(self.m_hostname, 
                port=self.port, 
                keepalive=self.keepAlive)

            self.state = MqttClient.Connecting
            self.m_client.loop_start()

    @QtCore.pyqtSlot()
    def disconnectFromHost(self):
        self.m_client.disconnect()

    def subscribe(self, path):
        if self.state == MqttClient.Connected:
            self.m_client.subscribe(path)

    #################################################################
    # callbacks
    def on_message(self, mqttc, obj, msg):
        mstr = msg.payload.decode("ascii")
        # print("on_message", mstr, obj, mqttc)
        self.messageSignal.emit(mstr)

    def on_connect(self, *args):
        # print("on_connect", args)
        self.state = MqttClient.Connected
        self.connected.emit()

    def on_disconnect(self, *args):
        # print("on_disconnect", args)
        self.state = MqttClient.Disconnected
        self.disconnected.emit()


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)

        lay = QtWidgets.QVBoxLayout(self)
        self.lcd_number = QtWidgets.QLCDNumber()
        lay.addWidget(self.lcd_number)

        self.client = MqttClient(self)
        self.client.stateChanged.connect(self.on_stateChanged)
        self.client.messageSignal.connect(self.on_messageSignal)

        self.client.hostname = "broker.hivemq.com"
        self.client.connectToHost()

    @QtCore.pyqtSlot(int)
    def on_stateChanged(self, state):
        if state == MqttClient.Connected:
            print(state)
            self.client.subscribe("shok2")

    @QtCore.pyqtSlot(str)
    def on_messageSignal(self, msg):
        try:
            val = float(msg)
            self.lcd_number.display(val)
        except ValueError:
            print("error: Not is number")


if __name__ == '__main__':
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

基于这个答案，我创建了一个库来将 paho 与 PyQt5 和 PySide2 一起使用：https://github.com/eyllanesc/qtmqtt

Answer 2

您是否必须创建方法 on_connect 使用订阅 on_connect 方法而不是运行。

def on_connect(self, client, userdata, flags, rc):
    print "Connected", rc
    client.subscribe("shok2")

def run(self):
    msg = subscribe.simple("shok2", hostname="broker.hivemq.com")
    print("%s %s" % (msg.topic, msg.payload))
    mqttc = mqtt.Client()
    mqttc.on_subscribe = self.on_subscribe
    mqttc.on_message = self.on_message
    mqttc.on_connect = on_connect
    mqttc.connect("broker.hivemq.com")
    mqttc.loop_forever()