1
驱动概览
这是什么

MQTT 驱动是 EasyPro 平台用于接入 MQTT 协议网关的驱动。网关下挂在各类传感器、执行器,通过 MQTT 协议将采集数据上报到平台的 Broker,平台订阅相关 Topic 接收数据、发布 Topic 下发控制指令。

与 Modbus 的「主站主动轮询」截然不同,MQTT 是事件驱动模型——设备有数据变化时主动推送,平台被动接收,实时性高、网络开销小。

关键事实
属性值说明
驱动名称mqttSPI 注册名
版本V1.0.0getDriverVersion()
接入模式EMBEDDED内嵌于平台主应用
协议库Eclipse Paho MQTT v3Java 标准 MQTT 客户端
通信模型发布/订阅(pub/sub)经 Broker 中转
Web 接口3 个查网关 / 绑网关 / ZIP 导入
定时任务1 个每 10 秒检测实例存活
实例属性校验无(空实现)配置由网关上报,非用户填写
2
发布订阅模型
三方通信拓扑

MQTT 通信不直接发生在平台与设备之间,而是经 Broker(消息代理)中转。发布者发消息到某个 Topic,Broker 分发给所有订阅了该 Topic 的接收者。

flowchart LR subgraph 设备侧 G1["网关 A"] G2["网关 B"] end B[" MQTT Broker
(EMQX)"] P[" EasyPro 平台
MQTT 驱动"] G1 -->|"发布 /data/up/..."| B G2 -->|"发布 /data/up/..."| B B -->|"推送订阅消息"| P P -.->|"发布 /set/down (控制)"| B B -.->|"转发"| G1 style B fill:#e8f0fe,stroke:#007aff,stroke-width:2px,color:#1c1c1e style P fill:#f0f0f5,stroke:#5856d6,stroke-width:2px,color:#1c1c1e style G1 fill:#e8f8f0,stroke:#34c759,color:#1c1c1e style G2 fill:#e8f8f0,stroke:#34c759,color:#1c1c1e
QoS 服务质量

MQTT 消息可指定 QoS(Quality of Service)等级,决定投递可靠性。本驱动对不同用途的消息采用不同 QoS:

QoS保证本驱动用途
0 至多一次不保证送达,开销最小心跳回复(高频,减负载)
1 至少一次保证送达,可能重复下行控制(setDown)(必须送达)
设计巧思

心跳回复用 QoS 0 是有意的——心跳消息量大且即使丢失下一次也会补上,用 QoS 0 可显著降低 Broker 与平台负载。

3
Topic 结构
订阅的 Topic(平台接收,7 个)
Topic用途
/create/up/+网关上线创建请求(首次接入)
/ins/up/open/+网关上报实例信息(设备类型/设备/点位)
/heartbeat/up/+网关心跳
/device/status/up/+设备状态变更
/data/up/+实时点位数据(核心)
/history/data/up/+历史数据补传
/reply/identifier/device/set/down设备对下行控制的回复

+ 是 MQTT 单层通配符,匹配任意网关 ID。

发布的 Topic(平台发送,4 个模板)
Topic 模板用途QoS
/{0}/reply/create/up回复网关创建(建立连接标志)—
/{0}/reply/heartbeat/up回复心跳0
/{0}/reply/ins/up/open回复实例信息(初始化结束标志)—
/{0}/identifier/set/down下行点位控制1

{0} 替换为具体网关 ID(gatewayId)。

4
整体架构
单客户端 + 多处理器管线

驱动只维护一个共享的 Paho MQTT 客户端连接到 Broker,订阅全部 7 个 Topic。每条入站消息由 MqttMessagePipLine 按 Topic 分发给对应的 handler,每个 handler 在独立虚拟线程上处理,互不阻塞。

flowchart TB B[" MQTT Broker"] -->|"消息到达"| CB["OnMessageCallback
messageArrived"] CB -->|"每条消息起虚拟线程"| PL["MqttMessagePipLine"] PL --> H1["DataUpHandler
实时数据"] PL --> H2["HeartbeatHandler
心跳/配置同步"] PL --> H3["DeviceStatusHandler
设备状态"] PL --> H4["HistoryDataUpHandler
历史数据"] PL --> H5["InitGatewayHandler
网关创建"] PL --> H6["InstanceInfoHandler
实例信息"] PL --> H7["ReplyIdentifierSetDown
下行回复"] style B fill:#e8f0fe,stroke:#007aff,stroke-width:2px,color:#1c1c1e style PL fill:#f0f0f5,stroke:#5856d6,stroke-width:2px,color:#1c1c1e
连接参数
参数值说明
cleanSessionfalse持久会话,断线重连后接收离线消息
connectionTimeout30 秒连接超时
automaticReconnecttrue自动重连
keepAliveInterval20 秒心跳间隔
maxInflight20同时飞行消息数
clientId配置值 + 随机后缀避免多实例 clientId 冲突
重连限流

connectionLost 回调对重连做了30 秒一次的限流,避免 Broker 故障时疯狂重连雪崩。

5
网关发现与导入
两种导入路径

网关接入平台有两条路径,最终都汇聚到同一套实例创建逻辑:

flowchart LR subgraph 在线["路径 A:在线自注册"] A1["网关发
/create/up"] --> A2["进入
noProjectGatewayCache"] A2 --> A3["用户调
chooseGateway"] end subgraph 离线["路径 B:离线 ZIP 导入"] B1["用户上传 ZIP
(JSON 配置)"] --> B2["调
importDriverInstance"] end A3 --> C["chooseNoProjectGateway
创建驱动实例"] B2 --> D["MqttImportUtil
9 阶段管线解析"] C --> E["网关上报 /ins/up/open
→ 物化设备类型/设备/点位"] D --> E style A1 fill:#e8f8f0,stroke:#34c759,color:#1c1c1e style B1 fill:#fff4e8,stroke:#ff9500,color:#1c1c1e style E fill:#e8f0fe,stroke:#007aff,stroke-width:2px,color:#1c1c1e
ZIP 导入 9 阶段管线

离线 ZIP 导入(MqttImportUtil.importMqttProjectByZipFile)是一个严谨的流水线,支持产品合并(同 productCode 合并为一个设备类型)与并发控制(Semaphore 10):

阶段动作
1. 校验检查 file、driverId 非空,并发导入互斥
2. 解压解压 ZIP 到临时目录
3. 解析逐个解析 JSON 配置文件
4. 点位容量检查点位数量是否超授权
5. 结构校验批量校验产品/设备/点位结构合法性
6. 合并冲突检查产品合并冲突
7. 开关校验switch-on 检查
8. 异步创建异步创建/更新设备类型、设备、点位
9. 返回结果返回成功/失败/冲突计数
6
心跳与配置同步
dataVersion 握手机制

网关每次心跳会带上自己的 dataVersion(配置版本号)。平台据此判断网关配置是否与平台一致,并通过 drq(data request)标志位要求网关上报最新配置:

sequenceDiagram participant G as 网关 participant H as HeartbeatHandler participant C as 平台缓存 G->>H: 心跳 (dataVersion=X) H->>C: 对比缓存 dataVersion alt 版本缺失或不一致 H->>G: 回复 drq_1(请上报新配置) Note over G: 网关开始上传 /ins/up/open else 正在上传中 H->>G: 回复 drq_2(继续) else 一致 H->>G: 回复 drq_0(无需操作) end
三个 drq 状态

drq_0=无需上报、drq_1=请求上报、drq_2=继续上报中。握手同步在 InstanceStorage.instanceIdMap 上加锁,避免与实例删除竞态。

7
下行控制
setDown 载荷结构

平台下发点位控制(issuePointValueBatch)时,按网关分组,构建 InstanceSetDown 载荷发布到 /{gwId}/identifier/set/down(QoS 1):

json
{
  "mId": "消息ID",
  "gwId": "网关ID",
  "da": [  // 数据数组
    {
      "productCode": "产品编码",
      "devices": [{
        "deviceCode": "设备编码",
        "identifiers": { "identifierCode[]": "value[]" }
      }]
    }
  ],
  "ps": { "scattered": "散点信息" }
}

值会按数据类型归一化:normalizeValueByDataType 将 bool 转 0/1,int/double/text 做类型强转。散点(uniqueCode 以 scattered 开头)有专门处理。

8
驱动与实例配置
驱动级配置(Broker 连接)

存在 iot_driver_config 表(非实例属性),由 MqttConfig 加载,定义平台如何连接 Broker:

配置项说明
urlBroker 连接地址(如 tcp://host:1883)
clientId客户端 ID 前缀(自动追加随机后缀)
username / passwordBroker 认证凭证
emqApiUrlEMQX HTTP API 地址(查网关 IP)
emqAppId / emqAppSecretEMQX API 认证
secret / expiresEMQX Token 密钥与过期时间
completionTimeout操作完成超时
实例级属性(每网关,9 项)

实例属性由网关上报自动填充,而非用户手工填写(故 validateDriverInstanceAttr 为空实现):

属性名来源说明
ipEMQX HTTP API 解析网关客户端 IP,重连时刷新
model网关上报/导入文件网关型号
version网关上报/导入文件固件版本
sn网关上报/导入文件序列号
gatewayId网关上报/导入文件网关 ID(= 实例名)
name网关上报网关显示名
passWord网关上报/导入文件接入密码
dataVersion网关上报配置版本(驱动 drq 握手)
mergeProduct用户/导入选择"1" 合并同名产品 / "0" 每网关独立
无点位属性 schema

MQTT 驱动没有固定的点位属性表——点位由网关上报的产品/设备/标识结构自动创建,配置即网关的物模型。

9
Web 接口
3 个 HTTP API
映射路径用途参数 / 上传
queryGateway查询已连接但未绑定项目的网关列表无参数;返回 GatewayLinkInfo[](含 IP)
chooseGateway将网关绑定到项目并创建实例gatewayId, password, projectId,可选 mergeProduct=true
importDriverInstance从 ZIP 批量导入网关项目ZIP 文件上传 + driverId,可选 mergeProduct

调用路径前缀:POST /driver/api/mqtt/{mapping}。

10
定时任务
实例存活检测
任务名Cron频率职责
MqttInstanceAliveTask0/10 * * * * ?每 10 秒检测全部实例状态:客户端掉线则重连;按 MqttCache.isOnline 设置 RUNNING/OFFLINE/CLOSED 并推送状态
11
常见问题
Q1
网关已上线,但平台查不到?
检查三处:① Broker 连接配置(url/username/password)是否正确;② 网关 clientId/topic 是否匹配订阅规则;③ 网关是否已发 /create/up 进入 noProjectGatewayCache。可用 queryGateway 接口确认。
Q2
导入 ZIP 后点位没创建?
ZIP 导入是异步创建点位(阶段 8),需等网关上报 /ins/up/open 后由 InstanceInfoHandler 物化。若网关在线则会很快完成;离线导入需等对应网关上线后触发配置同步(drq 握手)。
Q3
下行控制有时不生效?
下行用 QoS 1 保证送达,但需确认网关订阅了 /{gwId}/identifier/set/down。若网关离线,消息会暂存 Broker(cleanSession=false),网关重连后补投。检查 InstanceSetDown 载荷中 productCode/deviceCode/identifier 是否与网关物模型一致。
Q4
多个网关有相同产品,怎么避免重复建模?
导入/绑定时设 mergeProduct="1",相同 productCode 的产品会合并为一个设备类型,跨网关共享。设 "0" 则每个网关独立建设备类型(名称带 gatewayId 后缀)。