需求概述

主要實現(xiàn)一個流程：

1.采集Modbus電表數(shù)據(jù)上傳到MQTT指定topic

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

3.MQTT連接成功時發(fā)送注冊信息

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

5.云端下發(fā)控制EG8200繼電器并回復

對接資料

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

2.北向接口協(xié)議說明:

需求分析

注冊與連接工作

a.建立MQTT連接,發(fā)布到注冊主題（MQTT發(fā)布節(jié)點）注:MQTT訂閱發(fā)布節(jié)點連接信息共享，其他MQTT無需在配置連接參數(shù)，選擇第一次配置的連接參數(shù)即可

b.監(jiān)聽MQTT連接狀態(tài)(狀態(tài)變化節(jié)點)

c.封裝注冊信息(函數(shù)節(jié)點)

數(shù)據(jù)上報

a.modbus讀取電表數(shù)據(jù)（modbus讀節(jié)點）

b.判斷傳感器是否有回復并封裝上報報文(函數(shù)節(jié)點)

c.配置對應的發(fā)布主題(MQTT發(fā)布節(jié)點)

心跳推送

a.注入節(jié)點周期觸發(fā)(注入節(jié)點)

b.封裝心跳報文(函數(shù)節(jié)點)

c.MQTT發(fā)布數(shù)據(jù)（MQTT發(fā)布節(jié)點）

下發(fā)控制

a.MQTT訂閱云端下發(fā)主題（MQTT訂閱節(jié)點）

b.解析云端數(shù)據(jù)，并控制對應DO(函數(shù)節(jié)點)

c.響應MQTT控制情況（函數(shù)節(jié)點）

d.上報響應報文（MQTT發(fā)布）

實現(xiàn)流程框架

需求實現(xiàn)

1.監(jiān)聽MQTT狀態(tài)并發(fā)布注冊信息

a.從節(jié)點庫拖出一個MQTT發(fā)布節(jié)點,函數(shù)節(jié)點,調(diào)試節(jié)點,狀態(tài)監(jiān)測節(jié)點和條件判斷節(jié)點,MQTT發(fā)布用于消息發(fā)布,函數(shù)節(jié)點用于注冊信息封裝,狀態(tài)檢測節(jié)點用于檢測MQTT的連接狀態(tài)，調(diào)試節(jié)點用于打印發(fā)布的報文

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() // 秒
        // 有其他格式化字符需求可以繼續(xù)添加，必須轉(zhuǎn)化成字符串
    };
    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.數(shù)據(jù)上報

a.從節(jié)點庫拖出一個注入節(jié)點，函數(shù)節(jié)點和一個MQTT發(fā)布節(jié)點，注入節(jié)點用于周期觸發(fā)，函數(shù)節(jié)點用于封裝心跳報文，MQTT發(fā)布節(jié)點用于心跳發(fā)布

b.從上面可以看到日志窗口打印了數(shù)據(jù)報文，MQTTx也接收到了上報數(shù)據(jù)

c.數(shù)據(jù)上報流程使用的代碼塊：

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() // 秒
        // 有其他格式化字符需求可以繼續(xù)添加，必須轉(zhuǎn)化成字符串
    };
    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.從節(jié)點庫拿出一個注入節(jié)點,函數(shù)節(jié)點,MQTT發(fā)布節(jié)點。注入節(jié)點用于周期發(fā)發(fā)送心跳,函數(shù)節(jié)點用于封裝心跳報文,MQTT發(fā)布節(jié)點用于發(fā)布到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() // 秒
    // 有其他格式化字符需求可以繼續(xù)添加，必須轉(zhuǎn)化成字符串
  };
  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.下發(fā)控制及響應

a.從節(jié)點庫拿出一個MQTT訂閱節(jié)點，MQTT發(fā)布節(jié)點，函數(shù)節(jié)點和DO節(jié)點，MQTT訂閱節(jié)點和MQTT發(fā)布節(jié)點用于接收云端發(fā)來的數(shù)據(jù)和控制響應，函數(shù)節(jié)點用于解析下發(fā)的報文和控制的響應報文封裝

b.從上面可以看到日志窗口打印了報文，MQTTx也接收到了控制響應

c.控制流程使用的代碼塊

//云端下發(fā)解析
//############################
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() // 秒
        // 有其他格式化字符需求可以繼續(xù)添加，必須轉(zhuǎn)化成字符串
    };
    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;
//############################

通過以上步驟，你可以在邊緣網(wǎng)關上成功實現(xiàn) Modbus 協(xié)議到 MQTT 協(xié)議的轉(zhuǎn)換。這種集成方式不僅簡化了工業(yè)設備與物聯(lián)網(wǎng)平臺的通信，還增強了數(shù)據(jù)的實時性和可靠性。希望本文能為你提供有用的指導，助你更好地利用 Modbus 和 MQTT 技術

審核編輯 黃宇