使用 Paho Mqtt 客户端在 on_message 中发布不工作答案

【问题标题】：Publish in on_message with Paho Mqtt Client not working使用 Paho Mqtt 客户端在 on_message 中发布不工作
【发布时间】：2020-07-29 12:43:06
【问题描述】：

main 方法中的发布调用并不向代理发送消息，而是返回 (0,2)。 on_publish 不会被调用。我不知道我应该如何看待错误在哪里。我尝试了 qos=2 并运行主要方法 async 但它没有修复它。基本连接工作，因为 on_message 被调用。 mqtt-server 也适用于其他项目。有人有什么想法吗？

import paho.mqtt.client as mqtt
import subprocess
import shutil
import os
import glob
import logging
from datetime import datetime
import RPi.GPIO as GPIO
import time
from multiprocessing import Pool
GPIO.setmode(GPIO.BCM)

direction = 19
pwm = 26
GPIO.setup(direction, GPIO.OUT)
GPIO.setup(pwm, GPIO.OUT)

pwm = GPIO.PWM(pwm,19000)
speed = 0
pwm.start(speed)
running = False

client = mqtt.Client()
#First: pip3 install paho-mqtt
print("Starte Listener")

def start():
    GPIO.output(direction, GPIO.LOW)
    for dc in range(10, 40, 1):
        speed = dc 
        pwm.ChangeDutyCycle(speed)
        time.sleep(0.25)

def end():
    GPIO.output(direction, GPIO.LOW)
    for dc in list(reversed(range(0,40,1))):
        speed = dc 
        pwm.ChangeDutyCycle(speed)
        time.sleep(0.5)


def main(mqttClient):
    print("Started Spinning")
    running = True
    start()
    print(mqttClient.publish("scanner","shoot"))
    print("Waiting 124 seconds")
    time.sleep(124)
    print("ending spinning")
    end()
    print("finished spinning")
    running = False 

def on_connect2(client, userdata, flags, rc):
    client.subscribe("scanner",2)
    print("Connected "+str(rc))
    

# The callback for when a PUBLISH message is received from the server.
def on_message(mqttClient, userdata, msg):
    print("Recived message: "+str(msg.payload,'UTF-8'))
    if(str(msg.payload,'UTF-8') == "spin" and running == False):
        main(mqttClient)

def disconnected():
    print("Disconneted")

def on_publish(self, client, userdata, mid):
    print("onPublish")
    print(client,userdata,mid)

logger = logging.getLogger(__name__)
client.enable_logger(logger)

client.on_connect = on_connect2
client.on_message = on_message
client.on_publish = on_publish
client.on_disconnect = disconnected
client.username_pw_set("....","....")
client.connect("gx1", 1883,60)
client.loop_forever()

【问题讨论】：

如果发布返回 (0,2)，这意味着 0 - 发布成功和 2 - 分配的消息编号。不知道为什么 on_publish 不运行。从这里steves-internet-guide.com/publishing-messages-mqtt-client 。此外，“main”中的“running = True”看起来需要“全局运行”来更改全局变量，尽管这不会影响 on_publish 运行。

标签： python mqtt paho

【解决方案1】：

这是因为您阻塞了 MQTT 客户端线程。

当您调用client.loop_forever() 时，它会接管进程主线程并使用它来处理所有 MQTT 通信。当收到一条新消息时，MQTT 客户端线程会从网络堆栈中提取它并转换为消息对象，然后传递给on_message() 回调。该函数在客户端线程上运行。

当您致电 client.publish() 时，这将做两件事之一

如果消息是 QOS 0 并且小于网络 MTU，它将发布消息。
如果消息的 QOS 为 1 或 2 或大于网络 MTU，那么它将排队等待由客户端线程处理的消息。

您的代码中的问题* 是您将 main() 的返回阻止了 124 秒，这反过来又阻止了 on_message() 函数的返回，因此客户端线程无法发布您的消息。

如果你想做一些需要很长时间或阻塞on_message()（或任何回调函数）的事情，你应该启动一个单独的线程来运行它们。

*理论上您的消息看起来应该属于上面列出的情况 1，但可能还有其他因素导致它排队）

【讨论】：

【解决方案2】：

作为快速修复，我创建了一个新客户端：

def main(mqttClient):
    print("Started Spinning")
    running = True
    start()
    mq = mqtt.Client("shot_idicator")
    mq.username_pw_set("...","...")
    mq.connect("gx1", 1883,60)
    mq.publish("scanner","shoot")
    mq.disconnect()
    mq = None
    print("Waiting 124 seconds")
    time.sleep(124)
    print("ending spinning")
    end()
    print("finished spinning")
    running = False

现在可以了，但我认为这不是应该的。

【讨论】：