物联网：MQTT 与云

关键概念

MQTT 发布/订阅架构

理解 MQTT，关键是抓住"设备不直连、全走 Broker 中转"这一点：

搭建 Mosquitto Broker

先在树莓派上装 Broker（也可用第 10 章 Docker 版）。这里用 apt 直装：

# 安装 Mosquitto broker 和命令行客户端工具
sudo apt install -y mosquitto mosquitto-clients

# 设为开机自启并启动
sudo systemctl enable --now mosquitto

# 查看状态
systemctl status mosquitto

默认只允许本机连接。要让局域网其他设备也能连，需配置监听与认证：

# 编辑配置
sudo nano /etc/mosquitto/conf.d/local.conf

# 写入以下内容：
#   listener 1883 0.0.0.0     ← 监听所有网卡（局域网可访问）
#   allow_anonymous false     ← 禁止匿名，强制认证
#   password_file /etc/mosquitto/passwd

# 创建用户名密码（会提示输入密码）
sudo mosquitto_passwd -c /etc/mosquitto/passwd piuser

# 重启生效
sudo systemctl restart mosquitto

用命令行工具快速测试收发（开两个终端）：

# 终端1：订阅 home/# 下所有主题
mosquitto_sub -h localhost -u piuser -P 你的密码 -t "home/#" -v

# 终端2：发布一条温度消息
mosquitto_pub -h localhost -u piuser -P 你的密码 \
    -t "home/livingroom/temp" -m "26.5"

# 终端1 立即收到：home/livingroom/temp 26.5

Python 发布：上报传感器数据

用 paho-mqtt 库把第 5 章的传感器数据发布出去：

# 虚拟环境中安装 paho-mqtt
pip install paho-mqtt

# publisher.py — 定时发布温度数据
import paho.mqtt.client as mqtt
import json, time, random

BROKER = "localhost"       # Broker 地址
TOPIC = "home/livingroom/temp"

# 创建客户端（新版 paho 需指定 CallbackAPIVersion）
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.username_pw_set("piuser", "你的密码")

# 设置遗嘱：意外掉线时 Broker 替发 offline
client.will_set("home/livingroom/status", "offline", retain=True)

client.connect(BROKER, 1883, 60)
client.loop_start()          # 后台网络线程

# 上线，发布在线状态（保留消息）
client.publish("home/livingroom/status", "online", retain=True)

try:
    while True:
        temp = round(20 + random.random() * 10, 1)  # 模拟读数
        payload = json.dumps({"temp": temp, "ts": int(time.time())})
        # QoS 1 保证送达，retain 让新订阅者立即拿到最新值
        client.publish(TOPIC, payload, qos=1, retain=True)
        print(f"已发布: {payload}")
        time.sleep(5)
except KeyboardInterrupt:
    client.publish("home/livingroom/status", "offline", retain=True)
    client.loop_stop()
    client.disconnect()

Python 订阅：接收并响应

另一端订阅消息，收到后做处理（如超温报警、控制 GPIO）：

# subscriber.py — 订阅温度，超阈值报警
import paho.mqtt.client as mqtt
import json

# 连接成功回调：订阅主题
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"已连接 (code={reason_code})")
    client.subscribe("home/+/temp", qos=1)  # + 匹配所有房间

# 收到消息回调
def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    temp = data["temp"]
    print(f"[{msg.topic}] 温度 {temp}°C")
    if temp > 28:
        print("⚠️ 高温报警！可在此触发风扇/推送通知")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.username_pw_set("piuser", "你的密码")
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)
client.loop_forever()      # 阻塞循环，持续接收

连接云平台上报数据

数据除了本地流转，还可上云做长期存储、远程访问、图表分析。多数云 IoT 平台/物联网服务都兼容 MQTT，只需改 Broker 地址和凭证：

# cloud_report.py — 上报到云端 MQTT（以通用云 broker 为例）
import paho.mqtt.client as mqtt
import ssl, json, time

# 云平台提供的接入信息（各家不同，通常在控制台获取）
CLOUD_BROKER = "your-endpoint.iot.example.com"
CLOUD_PORT = 8883       # TLS 加密端口

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pi-livingroom")
client.username_pw_set("设备用户名", "设备密钥")

# 启用 TLS（云端必须加密，绝不能明文上公网）
client.tls_set(tls_version=ssl.PROTOCOL_TLS_CLIENT)

client.connect(CLOUD_BROKER, CLOUD_PORT, 60)
client.loop_start()

# 边缘计算：本地聚合后再上报（省流量）
readings = []
while True:
    readings.append(read_sensor())      # 每秒采一次
    if len(readings) >= 60:          # 攒够 60 条
        avg = sum(readings) / len(readings)
        # 只上报每分钟均值，而非每秒原始值 → 边缘计算
        client.publish("devices/pi/telemetry",
                       json.dumps({"avg_temp": round(avg, 2)}), qos=1)
        readings.clear()
    time.sleep(1)