邊緣網(wǎng)關(guān)教程-Modbus協(xié)議轉(zhuǎn)MQTT協(xié)議

需求概述

主要實(shí)現(xiàn)一個(gè)流程：

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

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

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

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

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

對(duì)接資料

1.南向接口 RS485對(duì)接協(xié)議(標(biāo)準(zhǔn)modbus)

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

需求分析

注冊(cè)與連接工作

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

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

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

數(shù)據(jù)上報(bào)

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

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

c.配置對(duì)應(yīng)的發(fā)布主題(MQTT發(fā)布節(jié)點(diǎn))

心跳推送

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

b.封裝心跳報(bào)文(函數(shù)節(jié)點(diǎn))

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

下發(fā)控制

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

b.解析云端數(shù)據(jù)，并控制對(duì)應(yīng)DO(函數(shù)節(jié)點(diǎn))

c.響應(yīng)MQTT控制情況（函數(shù)節(jié)點(diǎn)）

d.上報(bào)響應(yīng)報(bào)文（MQTT發(fā)布）

實(shí)現(xiàn)流程框架

需求實(shí)現(xiàn)

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

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

b.從上可以看見日志打印了注冊(cè)報(bào)文，以及MQTTx端以及接收到了注冊(cè)信息

c.注冊(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(), // 時(shí)
        "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ù)上報(bào)

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

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

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

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(), // 時(shí)
        "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é)點(diǎn)庫(kù)拿出一個(gè)注入節(jié)點(diǎn),函數(shù)節(jié)點(diǎn),MQTT發(fā)布節(jié)點(diǎn)。注入節(jié)點(diǎn)用于周期發(fā)發(fā)送心跳,函數(shù)節(jié)點(diǎn)用于封裝心跳報(bào)文,MQTT發(fā)布節(jié)點(diǎn)用于發(fā)布到MQTT

b.從上面可以看到日志窗口打印了心跳報(bào)文，MQTTx也接收到了心跳報(bào)

心跳上報(bào)流程使用的代碼塊：

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(), // 時(shí)
    "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ā)控制及響應(yīng)

a.從節(jié)點(diǎn)庫(kù)拿出一個(gè)MQTT訂閱節(jié)點(diǎn)，MQTT發(fā)布節(jié)點(diǎn)，函數(shù)節(jié)點(diǎn)和DO節(jié)點(diǎn)，MQTT訂閱節(jié)點(diǎn)和MQTT發(fā)布節(jié)點(diǎn)用于接收云端發(fā)來(lái)的數(shù)據(jù)和控制響應(yīng)，函數(shù)節(jié)點(diǎn)用于解析下發(fā)的報(bào)文和控制的響應(yīng)報(bào)文封裝

b.從上面可以看到日志窗口打印了報(bào)文，MQTTx也接收到了控制響應(yīng)

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;
}
//############################

//反饋?lái)憫?yīng)封裝
//############################
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(), // 時(shí)
        "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)關(guān)上成功實(shí)現(xiàn) Modbus 協(xié)議到 MQTT 協(xié)議的轉(zhuǎn)換。這種集成方式不僅簡(jiǎn)化了工業(yè)設(shè)備與物聯(lián)網(wǎng)平臺(tái)的通信，還增強(qiáng)了數(shù)據(jù)的實(shí)時(shí)性和可靠性。希望本文能為你提供有用的指導(dǎo)，助你更好地利用 Modbus 和 MQTT 技術(shù)

審核編輯 黃宇