MQTT 发布/订阅脚本手册

脚本语法为 Lua。MQTT 能力的整体概念（连接配置、发布配置、订阅配置的作用）请先阅读 MQTT 云边协同，本页详细说明发布/订阅脚本的写入参数、完整脚本示例与请求响应报文格式。

概述

可自定义 MQTT 数据发布格式，并配置 CRON 任务进行数据上传。
可自定义订阅消息解析，Edge-EMS Builder 后台会根据解析的结果，写入指定的点位。
可自定义订阅消息处理后，是否需要 response，以及 response 的 topic 和内容。

注：需要求使用者有编程基础，并熟悉 MQTT 订阅发布流程，具体脚本语法参见 Lua 脚本完整参考。

若脚本中引用的点位取不到值，会被注入哨兵字符串 __Null__，请显式判断 ${dataKey} == "__Null__"，不要假设其为空字符串或 nil。

MQTT 连接

参数名	参数值	是否必填	参数类型	描述说明
mqttConnectionConfigName	连接配置名称	是	String	MQTT 连接配置名称
status	Valid	是	String	状态(是否启用该连接)
protocol	mqtts	是	String	MQTT 协议(支持 mqtt/mqtts)
protocolVersion	3.1.1	是	String	MQTT 协议版本(支持 3.1/3.1.1/5.0)
serverHost	127.0.0.1	是	String	MQTT 连接地址
serverPort	8883	是	String	MQTT 连接端口
clientId	`return "demo-edge"`	是	String	返回客户端 ID 的 Lua 表达式
username	username	是	String	用户名
password	password	是	String	密码
cleanSession	true	是	Boolean	清理会话
sessionExpiryInterval	0	是	Number	会话过期间隔(MQTT5.0 协议特有)
keepAliveInterval	60	是	String	心跳间隔
connectTimeout	30	是	String	连接超时时间
sslEnable	true	是	Boolean	是否使用 SSL

MQTT 数据发布

参数名	参数值	是否必填	参数类型	描述说明
mqttConnectionConfigId	100101	是	Number	MQTT 连接配置 ID
mqttPublishConfigName	发布配置名称	是	String	MQTT 发布配置名称
topic	`return "demo/telemetry"`	是	String	返回主题的 Lua 表达式
qos	“1”	是	String	QoS 字符串枚举
publishStrategy	0/10 * * * * ?	是	String	发布策略(CRON 表达式)
publishContent	{“name”: “dataKey”}	是	String	发布内容(Lua 脚本)
note	note	是	String	备注

发布链路会将脚本结果序列化为 JSON。发布对象时直接返回 Lua table；提前调用 toJsonString 会形成带引号的 JSON 字符串。

publishContent 示例 1（构造 Table）

local map = {}
-- 时间
map.time = OffsetDateTime.now():unixMilli()
-- 普通数据值
map.name = "${数据8}"
-- 字符串数组
map.strList = { "${数据3}", "${数据4}" }
-- 数字数组
local d3 = ${数据3}
local d4 = ${数据4}
map.numberList = { d3, d4 }

return map

publishContent 示例 2（字符串拼接）

local json = '{"strList":["${数据3}","${数据4}"],"name":"${数据8}","time":' .. OffsetDateTime.now():unixMilli() .. ',"numberList":[' .. ${数据3} .. ',' .. ${数据4} .. ']}'
return jsonToMap(json)

解析结果

{
    "strList": [
        "3.6",
        "0.6"
    ],
    "name": "2",
    "time": 1742180870064,
    "numberList": [
        3.6,
        0.6
    ]
}

MQTT 消息订阅

参数名	参数值	是否必填	参数类型	描述说明
mqttConnectionConfigId	2	是	Number	MQTT 连接配置 ID
mqttSubscribeConfigName	订阅配置名称	是	String	MQTT 订阅配置名称
status	Valid	是	String	状态
topic	`return "demo/commands"`	是	String	返回主题的 Lua 表达式
qos	“1”	是	String	QoS 字符串枚举
actionContent	{}	是	String	动作内容(Lua 脚本)
needResponse	true	是	Boolean	是否需要 response
responseTopic	`return "demo/replies"`	启用响应时	String	返回响应主题的 Lua 表达式
response	`return {ok=true}`	启用响应时	String	响应脚本：字符串原样发送，table 序列化为 JSON
note	note	是	String	备注

actionContent：解析订阅的 message

写入参数

参数名	参数值	是否必填	参数类型	描述说明
dataType	Customization	是	String	要操作的数据类型：Customization/Collection/Aggregation
dataKey	功率	是	String	数据唯一 dataKey
dataValue	topic	是	String	数据值
availableDate	2023-01-15T07:00:00+08:00	否	String	生效日期（Customization）
expiredDate	2023-01-15T10:00:00+08:00	否	String	失效日期（Customization）

固定参数：

${subscribe_message}：从 topic 订阅到的 message

在使用 Edge-EMS Builder 时，请勿配置 dataKey 与上述固定配置 dataKey 重叠

actionContent 解析

local messageMap = jsonToMap(${subscribe_message})
local params = messageMap.params
local strategyList = jsonToListMap(toJsonString(params.strategy))
local date = params.date
local resultList = {}

for i, keyMap in ipairs(strategyList) do
    local availableDate = completeDateTime(date .. "T" .. keyMap.start)
    local expiredDate = completeDateTime(date .. "T" .. keyMap["end"])

    -- 要写入的数据类型：Collection（原始数据）/Customization（分时数据）/Aggregation（聚合数据）
    local modeMap = buildCommonWriteMap("Customization", "mode", tostring(keyMap.mode), availableDate, expiredDate, false)
    local chargeMap = buildCommonWriteMap("Customization", "charge", tostring(keyMap.charge), availableDate, expiredDate, false)
    local valueMap = buildCommonWriteMap("Customization", "value", tostring(keyMap.value), availableDate, expiredDate, false)

    table.insert(resultList, modeMap)
    table.insert(resultList, chargeMap)
    table.insert(resultList, valueMap)
end

table.insert(resultList, buildSlaveWriteMap("DeviceB", "Share", "数据4", 777, nil, nil, false))

return resultList

请求示例

/rpc/${iot_code}/${method}/request

订阅 Topic：/rpc/yanhua001/ems/strategy/request

注：iot_code 为聚合数据

发送消息

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

action 解析结果

解析结果要求：

dataType 只能是 Collection（原始数据）/Customization（分时数据）/Aggregation（聚合数据）
dataKey 必须存在
若同一个 Customization 的 dataKey 存在多个 value，必须保证每个 value 的生效区间不能重叠
在更新 Customization 数据时，会完全删除现有的，改为此次下发的数据

{
    "actionType": "SUBSCRIBE",
    "actionResult": true,
    "mqttActionContent": {
        "Customization": {
            "mode": [
                {
                    "dataType": "Customization",
                    "dataKey": "mode",
                    "dataValue": "1",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "mode",
                    "dataValue": "2",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ],
            "charge": [
                {
                    "dataType": "Customization",
                    "dataKey": "charge",
                    "dataValue": "true",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "charge",
                    "dataValue": "false",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ],
            "value": [
                {
                    "dataType": "Customization",
                    "dataKey": "value",
                    "dataValue": "100",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "value",
                    "dataValue": "100",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ]
        }
    },
    "mqttActionResult": {

    }
}

responseTopic：解析 response 订阅

固定参数：

${subscribe_message}：从 topic 订阅到的 message

${subscribe_result}：actionContent 执行是否成功

${subscribe_result_message}：actionContent 执行的结果，如 action 解析结果

在使用 Edge-EMS Builder 时，请勿配置 dataKey 与上述固定配置 dataKey 重叠

responseTopic 解析

local messageMap = jsonToMap(${subscribe_message})
local id = messageMap.id
local method = messageMap.method
return "/rpc/" .. "${iot_code}" .. "/" .. method .. "/response/" .. id

请求示例

/rpc/${iot_code}/${method}/response/${id}

目标响应 topic：/rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

发送消息

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

topic 解析结果

/rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

response：组装返回响应

response 解析

local messageMap = jsonToMap(${subscribe_message})
local id = messageMap.id
local jsonrpc = messageMap.jsonrpc
local result = "${subscribe_result}"

local resultMap = {}
resultMap.jsonrpc = jsonrpc
resultMap.result = result
resultMap.id = id

return resultMap

请求示例

/rpc/${iot_code}/${method}/response/${id}

目标响应 topic：/rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

发送消息

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

topic 解析结果

注：实际为压缩后的 json

{
    "result": "true",
    "id": "d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
    "jsonrpc": "2.0"
}

下一步

MQTT 云边协同的整体概念（连接/发布/订阅配置、断网续传）：MQTT 云边协同
完整的 Lua 内置函数与模块参考：Lua 脚本完整参考