EMQX 设备接入实战：MQTT 认证、数据上报与指令下发

模拟设备 ── telemetry / status ──> EMQX ──> MQTTX
模拟设备 <────── command ──────── EMQX <── MQTTX
模拟设备 ───── command-reply ───> EMQX ──> MQTTX

iot/demo/sensor-001/telemetry
iot/demo/sensor-001/status
iot/demo/sensor-001/command-reply

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "paho-mqtt>=2.1,<3"

$env:MQTT_HOST = "mqtt.example.com"
$env:MQTT_PORT = "8883"
$env:MQTT_USERNAME = "sensor-001"
# 私有 CA 才需要设置；使用受信任的公有 CA 时通常可以省略。
# $env:MQTT_CA = "C:\certs\ca.pem"

.\.venv\Scripts\python.exe device_demo.py

python3 -m venv .venv
.venv/bin/python -m pip install 'paho-mqtt>=2.1,<3'

export MQTT_HOST=mqtt.example.com
export MQTT_PORT=8883
export MQTT_USERNAME=sensor-001
# export MQTT_CA=/path/to/ca.pem

.venv/bin/python device_demo.py

"""EMQX tutorial: simulated sensor, TLS, Paho MQTT 2.x callback API."""
import getpass
import json
import os
import random
import ssl
import threading
import time
import uuid

import paho.mqtt.client as mqtt

DEVICE_ID = "sensor-001"
BASE = f"iot/demo/{DEVICE_ID}"
READY = threading.Event()
STATE = {"led": False}


def publish_json(client, suffix, data, retain=False):
    return client.publish(
        f"{BASE}/{suffix}", json.dumps(data), qos=1, retain=retain
    )


def on_connect(client, userdata, flags, reason_code, properties):
    READY.clear()
    if reason_code.is_failure:
        print("Connection rejected:", reason_code)
        return
    # Subscribe again after every successful reconnect.
    result, _ = client.subscribe(f"{BASE}/command", qos=1)
    if result != mqtt.MQTT_ERR_SUCCESS:
        print("Subscribe request failed:", result)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    if not reason_codes or any(code.is_failure for code in reason_codes):
        print("Command subscription denied:", reason_codes)
        client.disconnect()
        return
    publish_json(client, "status", {"online": True}, retain=True)
    READY.set()
    print("Device ready:", DEVICE_ID)


def on_disconnect(client, userdata, flags, reason_code, properties):
    READY.clear()
    print("Disconnected:", reason_code)


def on_message(client, userdata, message):
    if message.topic != f"{BASE}/command" or len(message.payload) > 4096:
        return
    try:
        command = json.loads(message.payload.decode("utf-8"))
        if not isinstance(command, dict):
            raise ValueError("Expected a JSON object")
        request_id = command.get("requestId")
        if not isinstance(request_id, str) or not request_id.strip():
            raise ValueError("Missing requestId")
        if command.get("action") != "set_led":
            raise ValueError("Unsupported action")
        if not isinstance(command.get("value"), bool):
            raise ValueError("value must be a boolean")
        # Simulation only: replace this assignment with a real device driver.
        STATE["led"] = command["value"]
        reply = {"requestId": request_id, "code": 0, "led": STATE["led"]}
        publish_json(client, "command-reply", reply)
        print("Command executed:", reply)
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
        print("Invalid command:", exc)


def main():
    host = os.environ["MQTT_HOST"]  # Hostname only, without mqtts://.
    password = os.getenv("MQTT_PASSWORD") or getpass.getpass("Device password: ")
    client = mqtt.Client(
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"device-{DEVICE_ID}",
        protocol=mqtt.MQTTv311,
        clean_session=True,
    )
    client.username_pw_set(os.getenv("MQTT_USERNAME", DEVICE_ID), password)
    client.tls_set_context(ssl.create_default_context(cafile=os.getenv("MQTT_CA")))
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.max_queued_messages_set(20)
    client.max_inflight_messages_set(10)
    client.will_set(f"{BASE}/status", '{"online": false}', qos=1, retain=True)
    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_disconnect = on_disconnect
    client.on_message = on_message
    client.connect_async(host, int(os.getenv("MQTT_PORT", "8883")), keepalive=60)
    client.loop_start()
    try:
        if not READY.wait(timeout=30):
            raise RuntimeError("Not ready: check network, TLS, authentication and ACL")
        while True:
            if READY.is_set():
                payload = {
                    "messageId": uuid.uuid4().hex,
                    "deviceId": DEVICE_ID,
                    "ts": int(time.time() * 1000),
                    "temperature": round(random.uniform(22, 28), 1),
                    "humidity": round(random.uniform(45, 65), 1),
                    "led": STATE["led"],
                }
                info = publish_json(client, "telemetry", payload)
                print("Telemetry queued:", info.rc, payload)
            time.sleep(10)
    except KeyboardInterrupt:
        print("Stopping device")
    finally:
        if client.is_connected():
            try:
                publish_json(client, "status", {"online": False}, True).wait_for_publish(2)
            except RuntimeError:
                pass
        client.disconnect()
        client.loop_stop()


if __name__ == "__main__":
    main()

{
  "requestId": "cmd-001",
  "action": "set_led",
  "value": true
}

{
  "requestId": "cmd-001",
  "code": 0,
  "led": true
}