把设备数据送到 EMQX，
剩下的交给 RoamLink。

00 / SCOPE

RoamLink 接收 MQTT，
不接管设备协议。

你的设备可以来自 BLE、Zigbee、Wi-Fi、Modbus、CAN、PLC、Home Assistant、Node-RED 或自研程序。RoamLink 不关心它如何读取物理设备，只要求最终有一个标准 MQTT 客户端把状态发布到 EMQX，并在需要控制时订阅命令 Topic。

01 / QUICK START

最短接入路径

1. 准备 Broker创建 EMQX Cloud Serverless，或者使用你自己的 EMQX；公网访问必须启用 TLS。
2. 规划命名空间为账号、空间、设备准备稳定 ID，例如 acme/rv01/fan01。
3. 创建两个账号设备/网关账号负责发布状态，RoamLink 账号负责订阅状态；需要控制时才允许发布命令。
4. 配置最小权限 ACL只允许两个账号访问这个空间的必要 Topic，最后增加默认拒绝规则。
5. 发布三类消息state、availability，以及可选的 command。
6. 用 MQTTX 验证先证明 Broker、TLS、账号、ACL 和消息都正确，再配置 RoamLink。
7. 在 RoamLink 映射选择界面模板，填写状态/命令 Topic，将 JSON 字段映射到组件。

02 / EMQX BROKER

准备 EMQX Broker

方案 A：EMQX Cloud Serverless

进入 EMQX Cloud 项目后创建 Serverless 部署。部署状态变为 Running 后，在 Overview 页面记录 Address，并下载 CA Certificate。Serverless 默认只开放加密连接：

官方参考：Serverless Connection Guide ↗ · Create a Serverless Deployment ↗

方案 B：自建 EMQX

自建 Broker 也可以使用，但要自行完成公网域名、证书、TLS Listener、WSS Listener、防火墙和升级维护。端口可以自定义；填入 RoamLink 的必须是外网可达的 WSS 地址。禁止把 EMQX Dashboard 管理端口直接当作 MQTT 端口。

03 / AUTH & ACL

账号分离与最小权限

不要让设备和 RoamLink 共用同一个 MQTT 账号，更不能使用 EMQX 控制台管理员凭据。推荐每个空间至少两个客户端身份：

EMQX Cloud 配置步骤

1. 打开 Access Control → Authentication，分别创建 device 与 roamlink 用户，使用随机且唯一的强密码。
2. 打开 Authorization → Username，按上表逐条添加 Allow。
3. 只读设备不创建任何 command Publish/Subscribe 权限。
4. 打开 All Users，最后添加 Topic #、Action Publish & Subscribe、Permission Deny。
5. 分别用两个账号测试：应允许的操作成功，越权 Topic 必须失败。

官方参考：Default Authorization ↗ · Authentication ↗

04 / TOPIC DESIGN

Topic 是长期接口，先规划再上线。

推荐使用四级资源前缀，再以消息用途结尾：

{account_id}/{space_id}/{device_id}/{message_type}

命名规则

• 只用小写英文字母、数字、短横线和下划线；不要在 ID 中使用 +、# 或 /。
• ID 一旦上线尽量不改；显示名称可以在 RoamLink 中单独修改。
• 不要把用户姓名、手机号、地址等隐私信息写入 Topic。
• 状态、在线状态、命令和事件分开，避免一个 Topic 同时承担多个生命周期。
• 命令 Topic 不使用 Retain，防止设备重连后执行旧命令。

05 / JSON PAYLOAD

状态消息应当可读、稳定、可扩展。

RoamLink 的映射层将 Topic 与 JSONPath 对应到界面能力。推荐每条状态都包含版本、时间、在线字段和值对象：

{
  "schema_version": 1,
  "device_id": "fan01",
  "timestamp": "2026-07-22T15:30:00+08:00",
  "online": true,
  "values": {
    "power": true,
    "actual_speed_percent": 48,
    "target_speed_percent": 50
  }
}

控制命令

{
  "request_id": "01J6RVFAN8K2",
  "action": "set",
  "values": { "power": true, "speed_percent": 50 },
  "issued_at": "2026-07-22T15:31:00+08:00",
  "expires_in_s": 10
}

设备执行后应在 state 或 event 中回传相同的 request_id 与执行结果。RoamLink 不能把“消息已发布”视为“物理设备已执行”。

06 / CONNECT YOUR SOURCE

选择你的设备接入方式

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>

WiFiClientSecure tls;
PubSubClient mqtt(tls);

const char* host = "YOUR_DEPLOYMENT.emqxsl.com";
const int port = 8883;
const char* stateTopic = "acme/rv01/fan01/state";
const char* availabilityTopic = "acme/rv01/fan01/availability";
const char* commandTopic = "acme/rv01/fan01/command";

void connectMqtt() {
  tls.setCACert(EMQX_CA_CERT);
  mqtt.setServer(host, port);
  mqtt.setCallback(onCommand);

  // availability/offline 是异常断线时由 Broker 发布的 LWT
  mqtt.connect(
    "acme-rv01-fan01",
    DEVICE_USERNAME,
    DEVICE_PASSWORD,
    availabilityTopic, 1, true, "offline"
  );

  mqtt.publish(availabilityTopic, "online", true);
  mqtt.subscribe(commandTopic, 1);
}

import json
import paho.mqtt.client as mqtt

HOST = "YOUR_DEPLOYMENT.emqxsl.com"
BASE = "acme/rv01/fan01"

client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    client_id="acme-rv01-fan01",
    protocol=mqtt.MQTTv5,
)
client.username_pw_set("acme-rv01-device", "DEVICE_PASSWORD")
client.tls_set(ca_certs="emqxsl-ca.crt")
client.will_set(f"{BASE}/availability", "offline", qos=1, retain=True)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        client.publish(f"{BASE}/availability", "online", qos=1, retain=True)
        client.subscribe(f"{BASE}/command", qos=1)

def on_message(client, userdata, msg):
    command = json.loads(msg.payload)
    # 在这里校验 action、参数范围、过期时间，再操作设备

client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=120)
client.connect(HOST, 8883, keepalive=30)

state = {"schema_version": 1, "online": True,
         "values": {"power": False, "actual_speed_percent": 0}}
client.publish(f"{BASE}/state", json.dumps(state), qos=1, retain=True)
client.loop_forever(retry_first_connection=True)

07 / VERIFY WITH MQTTX

先验证 MQTT，再配置 RoamLink。

1. 建立 device 连接使用 mqtts、8883、CA、device 凭据；发布 availability 和 state。
2. 建立 roamlink 测试连接使用另一个 Client ID 和 roamlink 凭据，订阅空间前缀。
3. 确认即时状态新订阅者应立即收到 retained state 和 availability。
4. 确认控制测试客户端向 command 发布非 retained 命令，device 客户端应收到。
5. 确认越权失败尝试发布不在 ACL 内的 state 或其他空间 Topic，Broker 必须拒绝。
6. 确认离线直接断网或终止 device 进程，订阅端应收到 retained offline LWT。

EMQX Cloud 官方推荐使用 MQTTX 先验证部署连接。调试完成后，不要在截图、日志或工单中保留真实密码。

08 / MAP INTO ROAMLINK

在 RoamLink 中添加 MQTT 设备

控制中心的自助映射向导正在建设。正式流程会包含以下字段；当前房车空间已经用相同模型手工配置：

09 / RELIABILITY

让在线状态与控制结果可信

• 每个客户端使用唯一 Client ID，否则新连接可能踢掉旧连接。
• 启用自动重连与指数退避，避免断网时高频重试。
• 状态中携带时间，RoamLink 才能区分“最后已知值”和“实时值”。
• 设备收到命令后检查参数范围、过期时间和幂等 request_id。
• 涉及加热、门锁、水泵等物理风险的设备必须由设备端实施安全限制。

10 / TROUBLESHOOTING

常见故障排查

11 / SECURITY

上线前安全清单

官方参考：EMQX Security Checklist ↗

12 / COMPLETE EXAMPLE

一台可控风扇的完整接口

发布状态

Topic: acme/rv01/fan01/state
QoS: 1
Retain: true

{
  "schema_version": 1,
  "device_id": "fan01",
  "timestamp": "2026-07-22T15:30:00+08:00",
  "online": true,
  "values": {
    "power": true,
    "actual_speed_percent": 48,
    "target_speed_percent": 50
  }
}

发布在线状态

Topic: acme/rv01/fan01/availability
Online payload: online
Last Will payload: offline
QoS: 1
Retain: true

接收控制命令

Topic: acme/rv01/fan01/command
QoS: 1
Retain: false

{
  "request_id": "01J6RVFAN8K2",
  "action": "set",
  "values": { "power": true, "speed_percent": 50 },
  "expires_in_s": 10
}

最终验收

• 设备断电后 1 个 Keep Alive 周期内显示 offline
• 新客户端订阅后立即收到最后 state
• RoamLink 账号不能发布 state
• 设备账号不能读取其他空间
• 旧 command 不会在重连后执行
• 执行结果带原 request_id 回传