使用 Paho MQTT 在 Python 中订阅 MQTT 时出现线程问题答案

【问题标题】：Thread issue while subscribing to MQTT in Python using Paho MQTT使用 Paho MQTT 在 Python 中订阅 MQTT 时出现线程问题
【发布时间】：2015-06-15 03:18:36
【问题描述】：

我有一个 python 程序，它监听 MQTT 主题并需要处理消息。我从命令行指定了一些参数来以不同的方式评估消息。

import argparse
import datetime
import json

import paho.mqtt.client as mqtt

### MQTT Functions
def on_connect(mqttc, obj, flags, rc):
    print("Connected! - " + str(rc))

def on_message(mqttc, obj, msg):
    print(msg.topic+" "+str(msg.qos)+" "+str(msg.payload))

def on_publish(mqttc, obj, mid):
    print("Published! "+str(mid))

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

def on_log(mqttc, obj, level, string):
    print(string)

if __name__ == "__main__":
    # Handle args
    parser = argparse.ArgumentParser(
        description='This is to be usedin conjunction with the WifiScanner on a Raspberry Pi')
    parser.add_argument('--topic', metavar='base/sub', type=str, nargs='?', help='Full topic to listen to. (Example "proximity/sensor")', default="proximity/#")
    parser.add_argument('--host', metavar='url', type=str, nargs='?',
                        help='UQL of MQTT server.')
    parser.add_argument('--graph', metavar='True/False', type=bool, nargs='?', help='Whether to print the data.', default=True)
    parser.add_argument('--timeout', metavar='sec', type=int, nargs='?', help='How long the device will be remembered', default=10)
    args = parser.parse_args()
    # MQTT
    mqttc = mqtt.Client()
    # mqttc.on_message = on_message
    mqttc.on_connect = on_connect
    mqttc.on_publish = on_publish
    mqttc.on_subscribe = on_subscribe
    # Uncomment to enable debug messages
    #mqttc.on_log = on_log
    mqttc.connect(args.host, 1883, 60)
    mqttc.subscribe(args.topic, 0)
    # Start to listen    
    while True:
        print mqttc.loop()

问题在于，我看不到将命令行参数传递给on_message 回调的简单方法。所以我尝试使用.loop的返回值。但是，当我尝试使用 Ctrl+Z 退出时（只有键盘中断有效），它不会退出 MQTT 线程并让它们继续运行。

文档和示例没有关于如何处理on_message 回调之外的消息以及如何干净退出的示例。

因此，任何解决此问题的帮助将不胜感激。

提前致谢

【问题讨论】：

使它们成为一个类的类方法并在该类上设置.args。

标签： python multithreading mqtt

【解决方案1】：

您可以使用Client() 构造函数中的userdata 参数。它最终被传递给每个回调。

【讨论】：

【解决方案2】：

已解决，谢谢James Mills

将回调放在一个类中：

class Receiver:
    def __init__(self, graph, timeout):
        self.graph = graph
        self.timeout = timeout

    def on_connect(self, mqttc, obj, flags, rc):
        print("Connected! - " + str(rc))

    def on_message(self, mqttc, obj, msg):
        print(msg.topic+" "+str(msg.qos)+" "+str(msg.payload))

    def on_publish(self, mqttc, obj, mid):
        print("Published! "+str(mid))

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

    def on_log(self, mqttc, obj, level, string):
        print(string)

然后像这样设置

mqttc.on_message = receiver.on_message

【讨论】：

【解决方案3】：

这里是在 paho 中使用类的另一个例子。

import paho.mqtt.client as mqtt
class client1:

    def on_connect(self, master, obj, flags, rc):
        self.master.subscribe('/temperature123')


    def on_message(self, master, obj, msg):
        print(str(msg.payload))

    def __init__(self,master):
        self.master=master
        self.master.on_connect=self.on_connect
        self.master.on_message=self.on_message
        self.master.connect("test.mosquitto.org",1883,60)

client=mqtt.Client()
ob1=client1(client)
client.loop_forever()

【讨论】：