MQTT客户端基于传入消息条件发布

Question

我正在用python paho mqtt库和一个mqtt客户端移动应用程序与test.蚊子.org服务器/代理进行测试。

下面的基本脚本连接到测试。我可以从移动mqtt客户端应用程序向此脚本发布消息，该脚本还可以通过def publish(client):函数每隔20秒向移动应用程序发布一条测试消息。

import random
import time
from paho.mqtt import client as mqtt_client

broker = 'test.mosquitto.org'
port = 1883 

# generate client ID with pub prefix randomly
client_id = "test_1"
topic_to_publish = f"laptop/publish"
topic_to_listen = f"mobile/publish"
topic_to_wildcard = f"testing/*"
username = ""
password = ""

def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
client.subscribe(topic_to_listen)
print(f"Connected to MQTT Broker on topic: {topic_to_wildcard}")
else:
print("Failed to connect, return code %dn", rc)
client = mqtt_client.Client(client_id)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.connect(broker, port)
client.on_connect = on_connect  # Define callback function for successful connection
client.on_message = on_message  # Define callback function for receipt of a message
return client

def publish(client):
msg_count = 0
while True:
time.sleep(20)
msg = f"hello from {client_id}: {msg_count}"
result = client.publish(topic_to_publish, msg)

# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send {msg} to topic {topic_to_publish}")
else:
print(f"Failed to send message to topic {topic_to_publish}")
msg_count += 1

def on_message(client, userdata, msg):  # The callback for when a PUBLISH message is received from the server.
print("Message received-> " + msg.topic + " " + str(msg.payload))

def run():
client = connect_mqtt()
client.loop_start()
publish(client)

if __name__ == '__main__':
run()

有人能给我一个关于如何修改def publish(client):功能的提示，而不是每20秒发射消息的while loop，但只有在从移动应用程序收到的消息等于字符串"zone temps"时才发布?

我是否正在从主要run函数中删除publish(client)以及从def publish(client):中删除while循环?谢谢任何提示，非常感谢。我遇到的是，当我运行这个修改版本时，我错过了一些东西，两者之间根本没有消息交换。

def on_message(client, userdata, msg):  
print("Message received-> " + msg.topic + " " + str(msg.payload))

if str(msg.payload) == "zone temps":
publish(client,"avg=72.1;min=66.4;max=78.8")

def run():
client = connect_mqtt()
client.loop_start()

if __name__ == '__main__':
run()

Answer 1

我也是初学者;但是我要创建一个变量来决定它是发布还是监听，比如:

phoneAppListener = 0

和

if str(msg.payload) == "zone temps":

当我打印我的有效负载时，它看起来像:

b'payload'

首先，你需要分割你的有效负载，如:

tempMsgHolder = str(msg.payload).split("'")

。tempMsgHolder[1]是你的纯有效负载。

if tempMsgHolder[1] == "zone temps": phoneAppListener = 1

phoneAppListener值决定0为监听，1为发布。在你的发布循环中设置这个

phoneAppListener == 1: publish your message

import random
import time
import threading
from paho.mqtt import client as mqtt_client

class moduleDatas:
broker = ('test.mosquitto.org')
port = (1883)
# generate client ID with pub prefix randomly
client_id = "test_1"
topic_to_publish = f"laptop/publish"
topic_to_listen = f"mobile/publish"
topic_to_wildcard = f"testing/*"
username = ""
password = ""
# Create clients object:
# You can create mqtt client obj using same pattern. Client has different on_msg or ex. 
mqttClient_1 = mqtt_client.Client(moduleDatas.client_id) # You can create what ever you want to create a new thread
def mqttClientConnect():
mqttClient_1.connect(moduleDatas.broker[0], moduleDatas.port[0])
mqttClient_1.loop_start() # It creates daemon thread while your main thread running, this will handle your mqtt connection.
@mqttClient_1.connect_callback()
def on_connect(client, userdata, flags, rc):
if rc == 0:
print(f"Connected to MQTT Broker on topic: {moduleDatas.topic_to_wildcard}")
else:
print("Failed to connect, return code %dn", rc)
@mqttClient_1.publish_callback()
def on_publish(client, userdata, mid):
print(mid) # If publish is success its return 1 || If mid = 1 publish success. || You can check your publish msg if it return failed try to send again or check your connection.
@mqttClient_1.message_callback()
def on_message(client, userdata, message):
temp_str = str(message.payload).split("'")
if temp_str[1] == "zone temps":
msg = "hello world" # <-- Your message here. Some func return or simple texts
mqttClient_1.publish(topic= moduleDatas.topic_to_publish, payload= msg, qos= 0)
def mqttClientSubscribe():
mqttClient_1.subscribe(moduleDatas.topic_to_listen)
def threadMqttClient1():
mqttClientConnect()
mqttClientSubscribe()
def buildThreads():
threads= []
t = threading.Thread(target=threadMqttClient1(), daemon= True)
threads.append(t)
# You can create on same pattern and append threads list.
for t in threads:
t.start()
while True: # this will your main thread, you can create an operation, ill go with just idling.
pass
if __name__ == "__main__":
buildThreads()