需求概述

主要實現一個流程：

1.采集Modbus電表數據上傳到MQTT指定topic

2.如果傳感器未響應需要發送報警信息到指定topic

3.MQTT連接成功時發送注冊信息

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

5.云端下發控制EG8200繼電器并回復

對接資料

1.南向接口 RS485對接協議(標準modbus)

2.北向接口協議說明:

需求分析

注冊與連接工作

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 技術

審核編輯 黃宇