如何在 mqtt python 中同时使用订阅和发布？

Question

我正在使用发布到主题test/123 的设备，其中123 是设备的名称。我需要订阅该主题（并处理收到的消息）；此外，我还需要向同一主题发送一个词（test/123）。设备只关注这个话题。

如何按内容区分传入和传出？更准确地说，如何正确发送。在on_message 方法中，您需要执行此操作，或者您需要创建另一个方法，然后是如何在那里接收传入消息。从传入的消息中，我需要获取设备的名称，然后使用它。

import paho.mqtt.client as mqtt
import paho.mqtt.publish as publish
# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("/test/#")


# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))
    imei = msg.topic.split('test/')[1]
    data = msg.payload.decode()
    print(imei)
    print(data)
    publish(imei)
    

def publish(imei):
    client = mqtt.Client()
    user = 'test'
    passw = '1111'
    client.username_pw_set(user,passw)
    client.connect("localhost",1883)
    
    topic = '/test/'+ imei
    client.publish(topic,'hello')
    print('SEND')
    client.disconnect()
    

client = mqtt.Client()
user = 'test'
passw = '1111'
client.username_pw_set(user,passw)
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)

client.loop_forever()

Answer 1

MQTT 不以任何方式区分客户端，这意味着如果客户端订阅给定主题，它将收到关于该主题的所有消息，包括它自己发布的消息。因此，使用您当前的设计，您将始终收到您发布的消息以响应第一条消息，这将触发重新发送该消息。

MQTT 消息不包含有关谁发布消息的任何信息，除非您选择将其添加到有效负载中，因此您无法将传入消息识别为您刚刚发布的消息，这将导致消息循环风暴.

正确的解决方案是不要对 2 条消息使用相同的主题。

MQTT v5 有一个标志，可以作为建立连接的一部分传递 这可以防止将消息返回给发布它们的客户端。目前看来 Paho Python 库没有设置此标志的方法。

如果您使用 MQTT v3.1.1 和 mosquitto 或 RSMB MQTT 代理，则可以设置一个未记录的选项（这不是 MQTT 规范的一部分），该选项也将阻止返回消息。以下代码仅适用于我提到的 2 个代理。

import paho.mqtt.client as mqtt

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("test/#")


# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))
    imei = msg.topic.split('test/')[1]
    data = msg.payload.decode()
    print(imei)
    print(data)
    publish(client, imei)
    

def publish(client,imei):
    topic = 'test/'+ imei
    client.publish(topic,'hello')
    print('SEND')
    

client = mqtt.Client()
user = 'test'
passw = '1111'
client.username_pw_set(user,passw)
client.on_connect = on_connect
client.on_message = on_message
client.enable_bridge_mode()

client.connect("localhost", 1883, 60)

client.loop_forever()

附言不要以 / 开头的主题根据规范合法，它会破坏共享订阅等内容，并在主题树的开头添加一个额外的 null。