paho mqtt 未发布

Question

编写了一个每秒检查 gpio 状态的代码，如果新结果与前一个结果不匹配，它应该发布它。问题是它没有发布它，但如果你输入一个印刷品，那么一切都很清楚。有什么问题？

from argparse import ArgumentError
from multiprocessing.connection import Client
import paho.mqtt.client as mqtt
import paho.mqtt.publish as publish
import re
import subprocess
import time



def on_connect(client, userdata, flags, rc):
 if rc == 0:
    print("Connected with result code "+str(rc))
    values = dict()
    k = 0   
    
    while True:

        DIN4R=subprocess.run("gpioget `gpiofind \"DIN4\"`",shell=True,check=True, capture_output=True)
        DIN3R=subprocess.run("gpioget `gpiofind \"DIN3\"`",shell=True,check=True, capture_output=True)
        DIN2R=subprocess.run("gpioget `gpiofind \"DIN2\"`",shell=True,check=True, capture_output=True)

        arr= str(DIN4R.stdout + DIN3R.stdout + DIN2R.stdout)

        arrr = re.sub("[^0,^1]", "", arr)



        if k % 2 == 0:
            values['0'] = arrr
        else:
            values['1'] = arrr
        if k != 0:
            if values['1'] != values['0']:
             global arrrr
             arrrr=arrr
             print(arrrr)
             client.publish("test/5555result", arrrr)

        k+=1
        time.sleep(2)




def on_publish(client, userdata, result):            
 print("data published \n")
 pass


client = mqtt.Client()
client.on_connect = on_connect
client.on_publish = on_publish
client.connect("test.mosquitto.org", 1883, 60)





client.loop()

Answer 1

您不能在任何客户端回调中运行阻塞代码，因为它们会停止客户端网络循环的运行。

在这种情况下，while True: 回调中有一个 while True: 循环，这意味着它永远不会返回。这行不通。

另外client.loop() 仅触发客户端网络循环的单次迭代，您需要调用client.loop_start() 或client.loop_forever()。第一个在单独的线程上运行循环，后者在当前线程上循环。

Answer 2

我不确定为什么添加 print() 可以解决问题 - 但程序中存在与使用 client.loop() 函数有关的问题。

这被设计为定期调用，以允许 MQTT 代码运行、发布和接收消息。此外 - 通过在一个 paho 回调中放置一个永久循环 - 您还可以阻止 MQTT 代码运行，因为 on_connect() 回调永远不会返回。

与其使用on_connect() 回调来运行您的代码，不如在MQTT 回调之外创建您的gpio 循环，并使用一个标志来检测MQTT 连接何时建立。

因为您有一个每 2 秒延迟一次的循环，所以最好使用 client.start_loop() 函数，以便客户端将在单独的线程中启动 MQTT 处理循环。这样你就不用担心重复调用client.loop()了。

例如。

from argparse import ArgumentError
from multiprocessing.connection import Client
import paho.mqtt.client as mqtt
import paho.mqtt.publish as publish
import re
import subprocess
import time

is_connected = False 

def on_connect(client, userdata, flags, rc):
 
 global is_connected 

 if rc == 0:
    print("Connected with result code "+str(rc))
    is_connected = True 


def on_publish(client, userdata, result):            
 print("data published \n")
 pass


client = mqtt.Client()
client.on_connect = on_connect
client.on_publish = on_publish
client.connect("test.mosquitto.org", 1883, 60)
client.start_loop()

while( not is_connected ):
    print("Waiting for MQTT connection ...")
    time.sleep(1) 
    
values = dict()
k = 0   

while True:

    DIN4R=subprocess.run("gpioget `gpiofind \"DIN4\"`",shell=True,check=True, capture_output=True)
    DIN3R=subprocess.run("gpioget `gpiofind \"DIN3\"`",shell=True,check=True, capture_output=True)
    DIN2R=subprocess.run("gpioget `gpiofind \"DIN2\"`",shell=True,check=True, capture_output=True)

    arr= str(DIN4R.stdout + DIN3R.stdout + DIN2R.stdout)

    arrr = re.sub("[^0,^1]", "", arr)

    if k % 2 == 0:
        values['0'] = arrr
    else:
        values['1'] = arrr
    if k != 0:
        if values['1'] != values['0']:
            arrrr=arrr
            print(arrrr)
            client.publish("test/5555result", arrrr)

    k+=1
    time.sleep(2)