如何使用Python开发物联网？MQTT协议实践

pip install paho-mqtt

import paho.mqtt.client as mqtt
import time
import json
import random

# MQTT Broker 配置
BROKER_ADDRESS = "broker.hivemq.com"  # 这是一个免费的公共测试Broker
PORT = 1883
TOPIC = "iot/sensor/temperature"

def on_connect(client, userdata, flags, rc):
    """当客户端连接到Broker时调用"""
    if rc == 0:
        print("已成功连接到MQTT Broker!")
    else:
        print(f"连接失败，返回码: {rc}")

def on_publish(client, userdata, mid):
    """当消息成功发布时调用"""
    print(f"消息已发布，MID: {mid}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_publish = on_publish

try:
    client.connect(BROKER_ADDRESS, PORT, 60)
    client.loop_start()  # 在后台线程中处理网络流量、回调等

    print(f"开始模拟温度传感器，数据将发布到主题: {TOPIC}")
    while True:
        temperature = round(random.uniform(20.0, 30.0), 2)  # 模拟温度
        timestamp = int(time.time())
        sensor_data = {
            "device_id": "temp_sensor_001",
            "temperature": temperature,
            "unit": "Celsius",
            "timestamp": timestamp
        }

        payload = json.dumps(sensor_data)
        client.publish(TOPIC, payload, qos=1) # QoS 1 确保消息至少到达一次
        print(f"发布数据: {payload}")
        time.sleep(5) # 每5秒发布一次数据

except KeyboardInterrupt:
    print("\n程序终止，断开MQTT连接...")
    client.loop_stop()
    client.disconnect()
except Exception as e:
    print(f"发生错误: {e}")
    client.loop_stop()
    client.disconnect()

import paho.mqtt.client as mqtt
import json

# MQTT Broker 配置
BROKER_ADDRESS = "broker.hivemq.com"
PORT = 1883
TOPIC = "iot/sensor/temperature" # 订阅与发布者相同的主题

def on_connect(client, userdata, flags, rc):
    """当客户端连接到Broker时调用"""
    if rc == 0:
        print("已成功连接到MQTT Broker!")
        client.subscribe(TOPIC, qos=1) # 订阅主题
        print(f"已订阅主题: {TOPIC}")
    else:
        print(f"连接失败，返回码: {rc}")

def on_message(client, userdata, msg):
    """当收到来自Broker的消息时调用"""
    print(f"\n收到消息 - 主题: {msg.topic} | QoS: {msg.qos}")
    try:
        payload = msg.payload.decode('utf-8')
        sensor_data = json.loads(payload)
        print(f"设备ID: {sensor_data.get('device_id')}")
        print(f"温度: {sensor_data.get('temperature')} {sensor_data.get('unit')}")
        print(f"时间戳: {sensor_data.get('timestamp')}")

        # 简单逻辑：如果温度过高，发出警报
        if sensor_data.get('temperature') > 28.0:
            print("!!! 警告: 温度过高 !!!")

    except json.JSONDecodeError:
        print(f"无法解析JSON: {msg.payload}")
    except Exception as e:
        print(f"处理消息时发生错误: {e}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

try:
    client.connect(BROKER_ADDRESS, PORT, 60)
    client.loop_forever() # 阻塞式循环，处理网络流量和回调

except KeyboardInterrupt:
    print("\n程序终止，断开MQTT连接...")
    client.disconnect()
except Exception as e:
    print(f"发生错误: {e}")
    client.disconnect()