JiangShan-web/MQTT接口.md

在线离线接口

主题：

/productid/deviceid/status/post

原始http推送接口

http://121.41.45.13:8025/receive

abcdefghijkmlnopqrstuvwxyz

新的http推送接口

http://121.36.111.27:8080/bridge/get

消息格式：json格式，具体字段如下：（其中status取值：4：离线，3：在线）：

{"status":3,"isShadow":1,"rssi":-51}

onenet平台上下线信息

{msg={"dev_name":"1","at":1742873510747,"pid":"JYr2f72uSJ","type":2,"status":1}, signature=25BcZvEQ+kk10rfB9gLnnQ==, time=1742873510772, id=5614fc489df3433f9891d6d4c7c3e5e7, nonce=b1R7hjgp} "status":1上线 0下线

物模型推送接口： 将数据推送至FASTBEE端

/productid/deviceid/property/post

消息格式：json格式，具体字段如下：

[{ "id": "co2", "value": "1", "remark": "" }]

规则引擎仅作于设备状态更新

import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil;

Long productId = 136 // 固定 productId String sysTopic = "" // 系统主题 String sysPayload = "" // 系统数据格式 // 1. 获取原始内容 String payload = msgContext.getPayload() msgContext.logger.info("原始数据：" + payload)

// 2. ✅ 只提取 msg 内容 try { // ✅ 使用 Groovy 兼容的正则写法，提取 msg 内的数据 String msgData = (payload =~ /msg={(.+?)}/)[0][1] msgContext.logger.info("提取后的 msg 数据：" + msgData)

// ✅ 解析提取后的数据
JSONObject msgObj = JSONUtil.parseObj("{" + msgData + "}")

// ✅ 提取 dev_name 作为 serialNumber
String serialNumber = msgObj.getStr("dev_name", "unknown")

// ✅ 解析 status 并进行转换
Integer status = msgObj.getInt("status", -1)
Integer convertedStatus = (status == 0) ? 4 : (status == 1) ? 3 : status

// ✅ 构造转换后的 sysPayload
JSONObject newObj = new JSONObject()
newObj.put("status", convertedStatus)
newObj.put("isShadow", 1)
newObj.put("rssi", msgObj.getInt("rssi", -51))  // 默认 rssi = -51

sysPayload = newObj.toString()
sysTopic = "/" + productId + "/" + serialNumber + "/status/post"

} catch (Exception e) { msgContext.logger.error("数据解析失败：", e) sysPayload = "{}" }

// 3. 打印调试信息 msgContext.logger.info("新主题：" + sysTopic) msgContext.logger.info("新内容：" + sysPayload) msgContext.logger.info("NewTopic 长度: " + sysTopic.length());

// 4. 设置新的数据 msgContext.setTopic(sysTopic) msgContext.setPayload(sysPayload) msgContext.logger.info("输出：" + msgContext.getTopic()); // 执行Action动作参数(脚本由系统自动生成) msgContext.setData("mqttBridgeID", 8);

定制设备状态

import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil;

String sysTopic = "/special/device" // 固定主题 String sysPayload = "" // 系统数据格式

// 1. 获取原始内容 String payload = msgContext.getPayload() msgContext.logger.info(" 原始数据：" + payload)

// 2. ✅ 只提取 msg 内容 try { // ✅ 使用 Groovy 兼容的正则写法，提取 msg 内的数据 String msgData = (payload =~ /msg={(.+?)}/)[0][1] msgContext.logger.info(" 提取后的 msg 数据：" + msgData)

// ✅ 解析提取后的数据
JSONObject msgObj = JSONUtil.parseObj("{"  + msgData + "}")

// ✅ 提取 dev_name 作为 serialNumber 
String serialNumber = msgObj.getStr("dev_name",  "unknown")

// ✅ 解析 status 并进行转换 (0=上线,1=下线)
Integer status = msgObj.getInt("status",  -1)
Integer convertedStatus = (status == 0) ? 0 : (status == 1) ? 1 : status

// ✅ 构造转换后的 JSON payload 
JSONObject newObj = new JSONObject()
newObj.put("serialNumber",  serialNumber)
newObj.put("status",  convertedStatus)

sysPayload = newObj.toString() 

} catch (Exception e) { msgContext.logger.error(" 数据解析失败：", e) // 错误时返回默认 JSON JSONObject errorObj = new JSONObject() errorObj.put("serialNumber", "error") errorObj.put("status", -1) sysPayload = errorObj.toString() }

// 3. 打印调试信息 msgContext.logger.info(" 新主题：" + sysTopic) msgContext.logger.info(" 新内容：" + sysPayload)

// 4. 设置新的数据 msgContext.setTopic(sysTopic) msgContext.setPayload(sysPayload) msgContext.logger.info(" 输出：" + msgContext.getTopic()); // 执行Action动作参数(脚本由系统自动生成) msgContext.setData("mqttBridgeID", 8);

ONENET 物模型

物模型

{ msg = { "notifyType": "property", "productId": "ebm1xlK9QJ", "messageType": "notify", "data": { "id": "1", "params": { "time": { "time": 1747031766274, "value": 3 }, "press": { "time": 1747031766273, "value": 4 }, "remove": { "time": 1747031766273, "value": 2 } } }, "deviceName": "D1231XS9KI9X" }, signature = YoYfkfdyNiYfPmJPLzz5wg == , time = 1747031766259, id = e20abecf848c408fb381bc03a23ebfd7, nonce = xQLnbAzQ }

[ { "id": "params", "remark": "", "value": "3" }, { "id": "press", "remark": "", "value": "3" }, { "id": "remove", "remark": "", "value": "3" } ]

[ { "id": "press", "remark": "", "value": "3" } ]

{ "id": "123", "version": "1.0", "params": { "time": { "value": 2, "time": 1706673129818 }, "press": { "value": 3, "time": 1706673129818 }, "remove": { "value": 2 "time": 1747031766273 } } }

{ "id": "1234567890123", "version": "1.0", "params": { "press": { "value": 4 }, "remove": { "value": 2 }, "time": { "value": 3 } } }

mqttfx

物模型规则脚本

import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil;

Long productId = 136 String sysTopic = "" String sysPayload = "[]"

try { // 1. 获取原始payload String payload = msgContext.getPayload().trim()

// 2. 安全提取msg JSON（兼容各种格式）
int msgStart = payload.indexOf('"msg":')  + 6 
int jsonStart = payload.indexOf('{',  msgStart)
int jsonEnd = payload.lastIndexOf('}') 

String msgJsonStr = payload.substring(jsonStart,  jsonEnd + 1)
msgContext.logger.info(" 提取的msg JSON: " + msgJsonStr)

// 3. 解析JSON 
JSONObject msgObj = JSONUtil.parseObj(msgJsonStr) 

// 4. 获取设备名（确保字段名正确）
String serialNumber = msgObj.getStr("deviceName",  "unknown")

// 5. 处理物模型数据（完全兼容的遍历方式）
JSONObject data = msgObj.getJSONObject("data") 
JSONObject params = data.getJSONObject("params") 

JSONArray propertyArray = new JSONArray()
// 使用兼容的遍历方式替代each闭包 
Set<String> keys = params.keySet() 
keys.each  { key ->
    JSONObject valueObj = params.getJSONObject(key) 
    if (valueObj != null) {
        JSONObject prop = new JSONObject()
        prop.set("id",  key)
        prop.set("remark",  "")
        prop.set("value",  valueObj.getStr("value")) 
        propertyArray.add(prop) 
    }
}

// 6. 设置输出 
sysPayload = propertyArray.toString() 
sysTopic = "/${productId}/${serialNumber}/property/post"

} catch (Exception e) { msgContext.logger.error(" 处理异常: " + e.getMessage()) } finally { // 确保有效输出 msgContext.setTopic(sysTopic ?: "/${productId}/unknown/property/post") msgContext.setPayload(sysPayload ?: "[]") msgContext.setData("mqttBridgeID", 8)

// 验证日志 
msgContext.logger.info("===  最终结果验证 ===")
msgContext.logger.info(" 设备名: " + (sysTopic.split("/")[2]  ?: "null"))
msgContext.logger.info("Payload 内容: " + sysPayload)

}

2合1脚本

import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil;

// 公共配置 Long productId = 136 String sysTopic = "" String sysPayload = ""

try { // 1. 获取原始payload String payload = msgContext.getPayload().trim() msgContext.logger.info(" 原始数据：" + payload)

// 2. 提取msg内容
String msgContent = payload 
if (payload.startsWith("{"))  {
    // 尝试直接解析为JSON
    try {
        JSONObject payloadObj = JSONUtil.parseObj(payload) 
        if (payloadObj.containsKey("msg"))  {
            msgContent = payloadObj.getStr("msg") 
        }
    } catch (Exception e) {
        // 可能是msg={...}格式 
        if (payload.contains("msg={"))  {
            int start = payload.indexOf("msg={")  + 4
            int end = payload.lastIndexOf("}")  + 1
            msgContent = payload.substring(start,  end)
        }
    }
} else if (payload.contains("msg={"))  {
    // 处理msg={"key":"value"}格式
    int start = payload.indexOf("msg={")  + 4
    int end = payload.lastIndexOf("}")  + 1
    msgContent = payload.substring(start,  end)
} else if (payload.contains("msg="))  {
    // 处理msg=JSON格式（不带大括号）
    int start = payload.indexOf("msg=")  + 4 
    msgContent = payload.substring(start) 
}

// 3. 解析msg内容 
JSONObject parsedObj = JSONUtil.parseObj(msgContent) 
msgContext.logger.info(" 解析后的JSON对象：" + parsedObj)

// 4. 判断数据类型并处理 
if (parsedObj.containsKey("notifyType")  && "property".equals(parsedObj.getStr("notifyType")))  {
    // 物模型数据处理 ========================================= 
    msgContext.logger.info(" 检测到物模型数据格式")
    
    // 获取设备名 
    String serialNumber = parsedObj.getStr("deviceName",  "unknown")
    
    // 处理物模型数据
    JSONObject data = parsedObj.getJSONObject("data") 
    JSONObject params = data.getJSONObject("params") 
    
    JSONArray propertyArray = new JSONArray()
    params.keySet().each  { key ->
        JSONObject valueObj = params.getJSONObject(key) 
        if (valueObj != null) {
            JSONObject prop = new JSONObject()
            prop.set("id",  key)
            prop.set("remark",  "")
            prop.set("value",  valueObj.getStr("value")) 
            propertyArray.add(prop) 
        }
    }
    
    // 设置物模型输出 
    sysPayload = propertyArray.toString() 
    sysTopic = "/${productId}/${serialNumber}/property/post"
    
} else if (parsedObj.containsKey("dev_name")  || parsedObj.containsKey("status"))  {
    // 设备状态处理 =========================================== 
    msgContext.logger.info(" 检测到设备状态数据格式")
    
    // 提取设备信息 
    String serialNumber = parsedObj.getStr("dev_name",  "unknown")
    Integer status = parsedObj.getInt("status",  -1)
    Integer convertedStatus = (status == 0) ? 0 : (status == 1) ? 1 : status
    
    // 构造状态JSON 
    JSONObject statusObj = new JSONObject()
    statusObj.put("serialNumber",  serialNumber)
    statusObj.put("status",  convertedStatus)
    
    // 设置状态输出
    sysPayload = statusObj.toString() 
    sysTopic = "/special/device"
    
} else {
    throw new Exception("无法识别的数据格式")
}

} catch (Exception e) { msgContext.logger.error(" 处理异常: " + e.getMessage())

// 创建错误响应 
JSONObject errorObj = new JSONObject()
errorObj.put("error",  e.getMessage()) 
errorObj.put("originalData",  msgContext.getPayload()) 

sysPayload = errorObj.toString() 
sysTopic = "/error/unknown"

} finally { // 确保有效输出 msgContext.setTopic(sysTopic) msgContext.setPayload(sysPayload) msgContext.setData("mqttBridgeID", 8)

// 记录最终结果
msgContext.logger.info("===  处理结果 ===")
msgContext.logger.info(" 主题: " + sysTopic)
msgContext.logger.info("Payload:  " + sysPayload)

}

压力传感器的id

{msg={at=1747123666940, imei=868256050123875, type=1, ds_id=3323_0_5700, value=1538.0, dev_id=2415928615}, msg_signature=mpYC8EvR7lB14AzqteyjhA==, nonce=nPyzGpMV}

{"msg":{"at":1747142877978,"imei":"868256050017168","type":1,"ds_id":"3323_0_5700","value":-1.0,"dev_id":2442383559},"msg_signature":"MHjUOUPj0GGfpKpyBZ6BzQ==","nonce":"TdTKIcVq"}

上下线消息

{msg={at=1747126859912, login_type=10, imei=868256050123875, type=2, dev_id=2442383559, status=1}, msg_signature=Z/cRScM2TLOUeaptE1hllg==, nonce=BSy0fFIY}

{"msg":{"at":1747126859912,"login_type":10,"imei":"868256050123875","type":2,"dev_id":2442383559,"status":1},"msg_signature":"Z/cRScM2TLOUeaptE1hllg==","nonce":"BSy0fFIY"}

移动消息

{"msg":{"at":1747275785411,"imei":"868256050099802","type":1,"ds_id":"30100_0_6500","value":"01","dev_id":2442561775},"msg_signature":"2dRxxL4t9HDW1EmRwIGhfQ==","nonce":"hM31WR88"}

脚本

import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil;

// 默认 topic String sysTopic = "/special/device"; String sysPayload = "";

// 获取原始 payload String payload = msgContext.getPayload(); msgContext.logger.info("原始 payload：" + payload);

try { // 提取 msg 内容 def matcher = (payload =~ /"msg":{(.+?)}/); if (!matcher.find()) { throw new Exception("未找到 msg 字段"); }

String msgData = matcher[0][1];
msgContext.logger.info("提取出的 msg 内容：" + msgData);

// 将 msgData 转为 JSON 字符串（key=value → "key":"value"）
String jsonLike = msgData.replaceAll(/(\w+)=([^,}]+)/) { all, k, v ->
    "\"${k}\":\"${v}\""
};
jsonLike = "{${jsonLike}}";
msgContext.logger.info("转换为 JSON 格式字符串：" + jsonLike);

JSONObject msgObj = JSONUtil.parseObj(jsonLike);

// 判断是否是物模型数据（包含 ds_id 字段）
if (msgObj.containsKey("ds_id")) {
    String dsid = msgObj.getStr("ds_id");
    if ("3323_0_5700" != dsid) {
        msgContext.logger.info("不处理的 ds_id：" + dsid);
        return; // 直接跳过不处理
    }

    // ✅ 物模型处理
    String imei = msgObj.getStr("imei", "unknown");
    String valueStr = msgObj.getStr("value", "0");
    Double valueNum = NumberUtil.parseNumber(msgObj.getStr("value", "0")).doubleValue();

    JSONArray array = new JSONArray();
    JSONObject data = new JSONObject();
    data.put("id", "press");
    data.put("remark", "");
    data.put("value", valueNum);
    array.add(data);

    sysPayload = array.toString();
    sysTopic = "/138/${imei}/property/post";

} else {
    // ✅ 在线离线数据处理
    String imei = msgObj.getStr("imei", "unknown");
    Integer status = NumberUtil.parseInt(msgObj.getStr("status", "-1"));
    Integer loginType = NumberUtil.parseInt(msgObj.getStr("login_type", "-1"));
    Long timestamp = msgObj.getLong("at", 0);

    JSONObject newPayload = new JSONObject();
    newPayload.put("serialNumber", imei);
    newPayload.put("status", status);
    newPayload.put("loginType", loginType);
    newPayload.put("timestamp", timestamp);

    sysPayload = newPayload.toString();
    sysTopic = "/special/device";
}

} catch (Exception e) { msgContext.logger.error("解析出错：", e); JSONObject errorPayload = new JSONObject(); errorPayload.put("serialNumber", "error"); errorPayload.put("status", -1); sysPayload = errorPayload.toString(); sysTopic = "/special/device"; }

// 设置 topic 和 payload msgContext.setTopic(sysTopic); msgContext.setPayload(sysPayload);

// 打印调试信息 msgContext.logger.info("新主题：" + sysTopic); msgContext.logger.info("新内容：" + sysPayload); msgContext.logger.info("输出：" + msgContext.getTopic());

// 可选设置 // 执行Action动作参数(脚本由系统自动生成) msgContext.setData("mqttBridgeID", 7); // 执行Action动作参数(脚本由系统自动生成)