边缘网关教程-Modbus协议转MQTT协议

需求概述

主要实现一个流程：

1.采集Modbus电表数据上传到MQTT指定topic

2.如果传感器未响应需要发送报警信息到指定topic

3.MQTT连接成功时发送注册信息

4.周期推送心跳到指定topic

5.云端下发控制EG8200继电器并回复

对接资料

1.南向接口 RS485对接协议(标准modbus)

点位	字段	寄存器地址	解析方式
A相电压	voltageA	40001~40002	float32 ABCD
B相电压	voltageB	40003~40004	float32 ABCD
C相电压	voltageC	40005~40006	float32 ABCD
A相电压	currentA	40007~40008	float32 ABCD
B相电压	currentB	40009~40010	float32 ABCD
C相电压	currentC	40011~40012	float32 ABCD

2.北向接口协议说明:

MQTT连接信息
Broker		139.129.229.113
Clientid		TestClient
Username		82000000305E144F
Password		EG12345678
Topic报文格式
功能	Topic	数据流向	报文示例
信息注册	data/sg/${sn}/info	网关->平台	{ "sn":"82000000305E144F", "time":"2023-01-01 12:00:00" }
数据上报	data/sg/${sn}/report	网关->平台	{ "sn":"82000000305E144F", "data":{ "voltageA":0, "voltageB":0, "voltageC":0, "currentA":0, "currentB":0, "currentC":0 }, time:"2023-01-01 12:00:00" }
心跳	data/sg/${sn}/heart	网关->平台	{ "sn":"82000000305E144F", "data":{}, "message":"heart", "time":"2023-01-01 12:00:00" }
报警	data/sg/${sn}/warn	网关->平台	{ "sn":"82000000305E144F", "data":{}, "message":"offline", "time":"2023-01-01 12:00:00" }
平台控制	/data/sg/${sn}/request	平台->网关	{ "event_id":"HsUCigC4Jk", "data": { "parameter": "DO1", "value": 0 } }
控制回复	data/sg/${sn}/response	网关->平台	{ "event_id":"HsUCigC4Jk", "message":"OK", "data":{}, "time":"2023-01-01 12:00:00" }

需求分析

注册与连接工作

a.建立MQTT连接,发布到注册主题（MQTT发布节点）注:MQTT订阅发布节点连接信息共享，其他MQTT无需在配置连接参数，选择第一次配置的连接参数即可

b.监听MQTT连接状态(状态变化节点)

c.封装注册信息(函数节点)

数据上报

a.modbus读取电表数据（modbus读节点）

b.判断传感器是否有回复并封装上报报文(函数节点)

c.配置对应的发布主题(MQTT发布节点)

心跳推送

a.注入节点周期触发(注入节点)

b.封装心跳报文(函数节点)

c.MQTT发布数据（MQTT发布节点）

下发控制

a.MQTT订阅云端下发主题（MQTT订阅节点）

b.解析云端数据，并控制对应DO(函数节点)

c.响应MQTT控制情况（函数节点）

d.上报响应报文（MQTT发布）

实现流程框架

需求实现

1. 监听MQTT状态并发布注册信息

a.从节点库拖出一个MQTT发布节点,函数节点,调试节点,状态监测节点和条件判断节点,MQTT发布用于消息发布,函数节点用于注册信息封装,状态检测节点用于检测MQTT的连接状态，调试节点用于打印发布的报文

b.从上可以看见日志打印了注册报文，以及MQTTx端以及接收到了注册信息

c.注册流程使用的代码块：

 

const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())

let obj = {
    "sn": "82000000305E144F",
    "time": fmt
}

function dateFormat(fmt, timestamp) {
    let ret;
    const opt = {
        "Y+": timestamp.getFullYear().toString(), // 年
        "m+": (timestamp.getMonth() + 1).toString(),  // 月
        "d+": timestamp.getDate().toString(), // 日
        "H+": timestamp.getHours().toString(), // 时
        "M+": timestamp.getMinutes().toString(), // 分
        "S+": timestamp.getSeconds().toString() // 秒
        // 有其他格式化字符需求可以继续添加，必须转化成字符串
    };
    for (let k in opt) {
        ret = new RegExp("(" + k + ")").exec(fmt);
        if (ret) {
            fmt = fmt.replace(ret[1], (ret[1].length == 1) ? (opt[k]) : (opt[k].padStart(ret[1].length, "0")))
        };
    };
    return fmt;
}

msg.payload = JSON.stringify(obj)
return msg

 

2. 数据上报

a.从节点库拖出一个注入节点，函数节点和一个MQTT发布节点，注入节点用于周期触发，函数节点用于封装心跳报文，MQTT发布节点用于心跳发布

b.从上面可以看到日志窗口打印了数据报文，MQTTx也接收到了上报数据

c.数据上报流程使用的代码块：

 

function dateFormat(fmt, timestamp) {
    let ret;
    const opt = {
        "Y+": timestamp.getFullYear().toString(), // 年
        "m+": (timestamp.getMonth() + 1).toString(),  // 月
        "d+": timestamp.getDate().toString(), // 日
        "H+": timestamp.getHours().toString(), // 时
        "M+": timestamp.getMinutes().toString(), // 分
        "S+": timestamp.getSeconds().toString() // 秒
        // 有其他格式化字符需求可以继续添加，必须转化成字符串
    };
    for (let k in opt) {
        ret = new RegExp("(" + k + ")").exec(fmt);
        if (ret) {
            fmt = fmt.replace(ret[1], (ret[1].length == 1) ? (opt[k]) : (opt[k].padStart(ret[1].length, "0")))
        };
    };
    return fmt;
}

var flag = msg.status
var data = msg.payload
var payload
const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())

if (flag == "TIMEOUT") {
    payload = {
        "sn": "82000000305E144F",
        "data": {},
        "message": "offline",
        "time": fmt
    }
    msg.payload = JSON.stringify(payload)
    return [msg, null]
}
else if (flag == "OK") {
    payload = {
        sn: "02C00081275A574E",
        data,
        time: fmt
    }
    msg.payload = JSON.stringify(payload)
    return [null, msg]
}

 

3. 心跳推送

a.从节点库拿出一个注入节点,函数节点,MQTT发布节点。注入节点用于周期发发送心跳,函数节点用于封装心跳报文,MQTT发布节点用于发布到MQTT

b.从上面可以看到日志窗口打印了心跳报文，MQTTx也接收到了心跳报

心跳上报流程使用的代码块：

 

const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())

function dateFormat(fmt, timestamp) {
  let ret;
  const opt = {
    "Y+": timestamp.getFullYear().toString(), // 年
    "m+": (timestamp.getMonth() + 1).toString(),  // 月
    "d+": timestamp.getDate().toString(), // 日
    "H+": timestamp.getHours().toString(), // 时
    "M+": timestamp.getMinutes().toString(), // 分
    "S+": timestamp.getSeconds().toString() // 秒
    // 有其他格式化字符需求可以继续添加，必须转化成字符串
  };
  for (let k in opt) {
    ret = new RegExp("(" + k + ")").exec(fmt);
    if (ret) {
      fmt = fmt.replace(ret[1], (ret[1].length == 1) ? (opt[k]) : (opt[k].padStart(ret[1].length, "0")))
    };
  };
  return fmt;
}

let obj = {
  "sn": "82000000305E144F",
  "data": {},
  "message": "heart",
  "time": fmt
}
msg.payload = JSON.stringify(obj)
return msg;

 

4. 下发控制及响应

a.从节点库拿出一个MQTT订阅节点，MQTT发布节点，函数节点和DO节点，MQTT订阅节点和MQTT发布节点用于接收云端发来的数据和控制响应，函数节点用于解析下发的报文和控制的响应报文封装

b.从上面可以看到日志窗口打印了报文，MQTTx也接收到了控制响应

c.控制流程使用的代码块

 

//云端下发解析
//############################
if (msg.payload.length < 10) { return }

if (typeof (msg.payload) == "object") {
    var obj = msg.payload
    //信息获取
    let event_id = obj.event_id;
    global.set("event_id", event_id)
    let parameter = obj.data.parameter
    let value = obj.data.value
    let arr = []
    switch (parameter) {
        case "DO1":
            arr = value
            break
        case "DO2":
            arr = value
            break
    }

    msg.payload = arr


    return msg;
}
//############################

//反馈响应封装
//############################
let event_id = global.get("event_id")
const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())
function dateFormat(fmt, timestamp) {
    let ret;
    const opt = {
        "Y+": timestamp.getFullYear().toString(), // 年
        "m+": (timestamp.getMonth() + 1).toString(),  // 月
        "d+": timestamp.getDate().toString(), // 日
        "H+": timestamp.getHours().toString(), // 时
        "M+": timestamp.getMinutes().toString(), // 分
        "S+": timestamp.getSeconds().toString() // 秒
        // 有其他格式化字符需求可以继续添加，必须转化成字符串
    };
    for (let k in opt) {
        ret = new RegExp("(" + k + ")").exec(fmt);
        if (ret) {
            fmt = fmt.replace(ret[1], (ret[1].length == 1) ? (opt[k]) : (opt[k].padStart(ret[1].length, "0")))
        };
    };
    return fmt;
}

let obj = {
    "event_id": event_id,
    "message": "OK",
    "data": {},
    "time": fmt

}

msg.payload = JSON.stringify(obj);
return msg;
//############################

 

通过以上步骤，你可以在边缘网关上成功实现 Modbus 协议到 MQTT 协议的转换。这种集成方式不仅简化了工业设备与物联网平台的通信，还增强了数据的实时性和可靠性。希望本文能为你提供有用的指导，助你更好地利用 Modbus 和 MQTT 技术

审核编辑 黄宇