在物联网系统开发中，我们经常希望在本地复现测试环境的设备消息。如果让本地服务直接连接测试环境 MQTT Broker，开发、测试等多个环境会同时消费同一条设备消息，还可能同时向真实设备发送命令。

本文介绍一种更安全、可维护的方案：使用 Docker Compose 运行本地 EMQX，由本地 EMQX 主动连接测试 EMQX，只订阅并镜像设备上行消息。本地业务服务只连接本地 Broker，下行命令不回流测试环境。

一、问题背景

普通 MQTT 订阅是广播语义。假设开发、测试和本地环境都订阅：

/sys/#

网关上报一次开机事件，三个环境都会收到并处理。若事件会触发查询拓扑、同步点表或下发配置，就可能出现：

• 同一设备被并发发送多条相同命令；
• 多个环境重复写入各自或共享的数据库；
• 网关并发生成、上传同名文件；
• 请求由一个环境发起，回复却被另一个环境消费；
• 本地调试意外影响真实测试设备。

共享订阅可以让一条消息只交给组内一个客户端：

$share/rene-iot-backend/sys/#

但它只适合同一环境的服务集群。组内实例应共享数据库、Redis、Kafka 和业务配置。不能用同一个共享组混合开发、测试和生产环境，否则消息会随机落到某个环境，请求与回复也可能分离。

二、推荐架构

本文采用单向镜像：

测试设备
   │ 设备上行 /s/#
   ▼
测试 EMQX
   │ 本地 EMQX 作为 MQTT 客户端主动订阅
   ▼
本地 EMQX
   │ 保持原 Topic 重新发布
   ▼
本地 rene-iot / MQTTX

只允许下面的方向：

测试 EMQX  ──设备上行──>  本地 EMQX

禁止下面的方向：

本地 EMQX  ──设备下行──>  测试 EMQX

这种方向的好处是测试服务器不需要主动访问开发者电脑。本地 EMQX 只要能通过公司网络或 VPN 访问测试 Broker 即可。

三、使用 Docker Compose 启动本地 EMQX

创建目录并保存以下 docker-compose.yml：

services:
  emqx:
    image: ${EMQX_IMAGE:-emqx/emqx:5.8.4}
    container_name: rene-local-emqx
    restart: unless-stopped
    environment:
      TZ: ${TZ:-Asia/Shanghai}
      EMQX_DASHBOARD__DEFAULT_USERNAME: ${EMQX_DASHBOARD_USERNAME:-admin}
      EMQX_DASHBOARD__DEFAULT_PASSWORD: ${EMQX_DASHBOARD_PASSWORD:-LocalEmqx123!}
    ports:
      - "${EMQX_MQTT_PORT:-1883}:1883"
      - "${EMQX_WS_PORT:-8083}:8083"
      - "${EMQX_DASHBOARD_PORT:-18083}:18083"
    volumes:
      - emqx-data:/opt/emqx/data
      - emqx-log:/opt/emqx/log
    healthcheck:
      test: ["CMD", "/opt/emqx/bin/emqx_ctl", "status"]
      interval: 10s
      timeout: 5s
      retries: 12
      start_period: 20s

volumes:
  emqx-data:
    name: rene-local-emqx-data
  emqx-log:
    name: rene-local-emqx-log

建议再创建 .env，避免每次查找启动参数：

EMQX_IMAGE=emqx/emqx:5.8.4
TZ=Asia/Shanghai
EMQX_MQTT_PORT=1883
EMQX_WS_PORT=8083
EMQX_DASHBOARD_PORT=18083
EMQX_DASHBOARD_USERNAME=admin
EMQX_DASHBOARD_PASSWORD=请修改为本地强密码

启动：

docker compose up -d
docker compose ps
docker compose logs -f emqx

停止但保留配置和数据：

docker compose down

升级镜像后重建：

docker compose pull
docker compose up -d

Dashboard 地址：

http://127.0.0.1:18083

命名卷会保留账号、Connector、Source 和 Rule。不要随意执行下面的命令：

docker compose down -v

-v 会删除这些持久化数据。

四、在测试 EMQX 创建只读镜像账号

为本地镜像单独创建账号，例如：

用户名：local_mirror
密码：使用独立强密码

该账号只允许订阅设备上行 Topic：

/sys/+/+/+/s/#

明确禁止：

发布 /sys/#
订阅 /sys/+/+/+/c/#

不要复用网关账号、管理员账号或业务服务账号。ACL 是最后一道安全边界：即使本地 Connector 配错，镜像账号也不能向测试设备发布指令。

如果只调试一个网关，可以进一步缩小权限：

/sys/gw118/+/+/s/#

五、在本地 EMQX 创建远程 MQTT Source

登录本地 Dashboard，进入：

Integration → Rules → Create → Add Input → MQTT Broker

创建 Connector：

Name: test_emqx
Server: 测试EMQX地址:1883
Username: local_mirror
Password: 镜像账号密码
Client ID Prefix: rene-local-mirror
Clean Start: true

测试连接成功后创建 Source：

Name: test_device_upstream
Topic: /sys/+/+/+/s/#
QoS: 1

这里由本地 EMQX 主动连接测试 EMQX，因此本机需要能访问测试 Broker 的 1883 端口。如果测试环境只开放 TLS，则应使用对应 TLS 端口并配置 CA 证书。

六、把远程消息重新发布到本地 Broker

MQTT Source 收到的消息不会自动成为本地 Broker 消息，还需要在同一条 Rule 中添加：

Add Action → Republish

填写：

Topic: ${topic}
Payload: ${payload}
QoS: ${qos}
Retain: false

保持原 Topic 是为了让现有业务代码继续按原协议解析：

/sys/{gwSn}/{productKey}/{deviceSn}/s/...

不要在这条 Rule 中添加到测试 Broker 的 MQTT Sink。数据方向只能是远程 Source 到本地 Republish。

七、本地 rene-iot 连接本地 EMQX

本地启动 rene-iot 时使用：

EMQX_HOST=127.0.0.1
EMQX_PORT=1883
EMQX_TOPICS=/sys/#
EMQX_SERVER_USERNAME=本地EMQX业务账号
EMQX_SERVER_PASSWORD=本地EMQX业务账号密码

如果 rene-iot 本身也运行在另一个 Docker 容器中，容器里的 127.0.0.1 指向它自己，不能指向宿主机。Windows 和 macOS Docker Desktop 可以使用：

EMQX_HOST=host.docker.internal

若 rene-iot 与 EMQX 在同一个 Compose 网络中，则使用服务名：

EMQX_HOST=emqx

本地服务还应使用隔离的 MySQL、Redis、Kafka 和 FTP。只隔离 MQTT、不隔离存储，仍可能修改测试数据。

八、验证镜像

使用 MQTTX 连接本地 Broker：

Host: 127.0.0.1
Port: 1883
Topic: /sys/#

让测试网关产生一条上行消息，依次检查：

1. 测试 EMQX 能看到原始消息；
2. 本地 Rule 的命中计数增加；
3. MQTT Source 和 Republish Action 状态正常；
4. 本地 MQTTX 收到相同 Topic 和 Payload；
5. 本地 rene-iot 日志出现对应消息；
6. 测试设备没有收到任何来自本地的 /c/# 指令。

也可以查看容器状态：

docker compose ps
docker compose logs --tail 200 emqx

九、这个方案能测试什么

适合测试被动上行链路：

• 设备启动事件；
• 拓扑上报；
• 遥测、遥信和实时属性；
• 告警和运行事件；
• 测试环境原本产生的设备回复；
• 本地解析、缓存、规则计算和数据库写入。

不能完整测试需要本地主动下发的链路：

• 手动同步点表；
• 主动查询拓扑；
• 文件上传或下载命令；
• 策略下发和设备控制。

原因是这些操作需要本地发布 /c/#，而本文刻意禁止它回流测试设备。即使随后镜像到一条测试环境产生的回复，也不能保证请求 ID 与本地请求匹配。

如果必须完整测试主动命令，应准备专用测试网关，并只对该网关建立受控双向通道。例如只允许：

/sys/dev-gateway-001/+/+/c/#
/sys/dev-gateway-001/+/+/s/#

不要为整个 /sys/# 建立双向桥接。

十、常见问题排查

1. Connector 无法连接测试 EMQX

检查：

• 本机是否连接公司 VPN；
• 测试 Broker 地址和端口是否可达；
• 防火墙是否允许访问；
• 用户名和密码是否正确；
• 测试环境是否要求 TLS；
• 镜像账号是否被 ACL 拒绝。

2. Source 有数据，但本地订阅收不到

检查是否添加了 Republish Action，并确认：

Topic=${topic}
Payload=${payload}

MQTT Source 只是规则输入，不会自动把远程消息发布成本地 MQTT 消息。

3. 本地 rene-iot 收不到，本地 MQTTX 可以收到

检查：

• EMQX_HOST 是否确实指向本地 Broker；
• rene-iot 是否订阅 /sys/#；
• 本地 EMQX 业务账号是否具有订阅权限；
• Docker 容器内是否错误使用了 127.0.0.1；
• rene-iot MQTT clientId 是否与另一实例冲突。

4. 同一条消息在本地出现多次

检查：

• 是否创建了重复的 Source 或 Rule；
• 是否同时存在测试到本地和本地到测试的双向桥接；
• Republish 是否再次触发同一规则形成循环；
• 测试设备是否本身重复上报 QoS 1 消息。

5. 本地服务误向测试设备发指令

立即禁用 Connector/Rule，并检查测试环境镜像账号 ACL。安全性不能只依靠 Topic 过滤，镜像账号本身必须没有发布权限。

十一、共享订阅应该用在哪里

同一环境部署多个 rene-iot 实例时，可以使用：

$share/rene-iot-prod/sys/#

Broker 会把每条消息推送给组内一个在线客户端。这不是业务代码轮询，而是 MQTT Broker 负载均衡。

但共享订阅组内的服务必须共享相同的 Redis、数据库、Kafka 和配置。开发、测试、生产应该使用不同 Broker、Topic 命名空间或严格 ACL，不应加入同一个共享组。

十二、总结

本地调试真实设备数据时，最重要的不是“把消息复制过来”，而是建立清晰的数据方向和权限边界：

• 本地 EMQX 主动连接测试 EMQX；
• 只订阅设备上行 /s/#；
• 测试环境使用只读镜像账号；
• 本地保持原 Topic 重新发布；
• 本地业务只连接本地 Broker；
• 数据库、Redis、Kafka 和 FTP 同样隔离；
• 主动设备命令使用专用测试网关，不开放全局双向桥接。

这样既能在本地复现接近真实的设备消息，又不会让一次调试演变成对测试设备的意外控制。