Python 实现mqtt协议的上位机测试

下面是使用Python实现MQTT协议的上位机测试的示例代码，需要使用paho-mqtt库：

import paho.mqtt.client as mqtt

# 连接回调函数
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    # 订阅主题
    client.subscribe("test/topic")

# 消息接收回调函数
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

# 创建客户端对象
client = mqtt.Client()

# 设置连接回调函数
client.on_connect = on_connect

# 设置消息接收回调函数
client.on_message = on_message

# 连接MQTT服务器
client.connect("127.0.0.1", 1883, 60)

# 启动MQTT客户端
client.loop_start()

# 发布消息
client.publish("test/topic", "Hello, World!")

# 持续运行
while True:
    pass

需要注意的是，需要将127.0.0.1和1883替换为您实际使用的MQTT服务器的IP地址和端口号。此外，为了保持程序持续运行，可以在程序的最后添加一个无限循环。