如何在Python的MQTT中仅发送与上次不同的Topic控制指令？

我用Python编写了MQTT代码，通过电位计（数值范围0-1023）控制LED：当收到Medir主题的电位计数值后，会根据区间判断向Valve_OC主题发送对应控制指令。现在需要实现仅当当前要发送的指令与上次发送的指令不同时，才执行MQTT发布操作——比如电位计数值持续低于300时，不会重复发送Abrir,100指令。我了解可以用MQTT保留消息实现，但不清楚如何集成到现有代码中。

解决方案 #

这里提供两种实现方式，其中第二种结合了你提到的MQTT保留消息功能：

方式1：本地记录上次指令（简单直接） #

不需要依赖MQTT服务端，直接在代码中维护一个变量存储上次发布的指令，每次生成新指令后对比，仅当不同时执行发布操作。

方式2：结合MQTT保留消息（符合需求） #

保留消息会让MQTT服务器存储最后一条发布的消息，新订阅的客户端能立即获取到最新状态。我们可以通过保留消息初始化客户端的指令记录，同时本地对比避免重复发布。

修改后的完整代码

import time
import paho.mqtt.client as mqtt  # 补充缺失的MQTT库导入

# 全局变量：存储上次发布的指令，初始为空
last_published_command = ""

def on_connect(client, userdata, flags, rc):
    print("Conectando al Servidor - " + str(rc))
    client.subscribe("Valve_OC")
    client.subscribe("Medir")
    # 订阅后会自动接收Valve_OC的保留消息，触发on_message初始化last_published_command

def on_message(client, userdata, message):
    global last_published_command
    # 处理Valve_OC主题的消息（包括保留消息）
    if message.topic == "Valve_OC":
        last_published_command = str(message.payload.decode("utf-8"))
        print(f"更新上次指令记录：{last_published_command}")

def on_message_medir(client, userdata, message):
    global last_published_command
    time.sleep(1)
    message_medir = str(message.payload.decode("utf-8"))
    print("valor medido en arduino =", message_medir)
    valor_mensaje = int(float(message_medir))

    # 根据电位计数值生成对应指令
    current_command = ""
    if valor_mensaje <= 300:
        current_command = "Abrir,100"
        print("abrir completamente la valvula")
    elif 300 < valor_mensaje <= 600:
        current_command = "Abrir,60"
        print("abrir valvula al 60")
    elif 600 < valor_mensaje < 800:
        current_command = "Abrir,30"
        print("abrir al 30")
    elif valor_mensaje >= 800:
        current_command = "Cerrar"
        print("cerrar valvula")

    # 仅当当前指令与上次不同时才发布
    if current_command != last_published_command:
        # 发布时添加retain=True，让服务器保留这条消息
        client.publish("Valve_OC", current_command, retain=True)
        print(f"valor publicado en topic Valve_OC: {current_command}")
        # 更新上次发布的指令记录
        last_published_command = current_command
    else:
        print(f"指令未变化，无需重复发布：{current_command}")
    
    time.sleep(3)

def on_publish(client, obj, mid):
    print("Mensaje: " + str(mid))

def on_subscribe(client, obj, mid, granted_qos):
    print("Subscribed: " + str(mid) + " " + str(granted_qos))

def on_log(client, obj, level, string):
    print(f"Log: {string}")

if __name__ == "__main__":
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.on_message = on_message
    client.message_callback_add("Medir", on_message_medir)
    client.on_subscribe = on_subscribe
    client.on_log = on_log

    client.connect("test.mosquitto.org", 1883)
    client.loop_forever()

关键修改点说明

1. 补充MQTT库导入：原代码缺失import paho.mqtt.client as mqtt，添加后才能正常运行。
2. 全局变量记录指令：用last_published_command存储上次发布的指令，避免重复发送。
3. 保留消息配置：发布指令时添加retain=True参数，让MQTT服务器持久化存储最后一条指令。
4. 初始化指令记录：通过on_message回调接收Valve_OC的保留消息，确保客户端重启后能获取到最新的指令状态，避免重复发布。

内容的提问来源于stack exchange，提问作者Abdel Essi