Influxdb數(shù)據(jù)庫是什么？邊緣計算網(wǎng)關(guān)如何對接Influxdb數(shù)據(jù)庫？

Influxdb是非關(guān)系型數(shù)據(jù)庫中的時序數(shù)據(jù)庫，全稱時間序列數(shù)據(jù)庫數(shù)據(jù)庫。MySQL作為一種傳統(tǒng)的關(guān)系型數(shù)據(jù)庫，其擴展性相對有限，特別是在處理大規(guī)模并發(fā)讀寫時，其性能可能會受到較大的影響。而時序數(shù)據(jù)庫則通過分布式架構(gòu)設(shè)計，能夠更好地支持大規(guī)模并發(fā)讀寫，具有更好的擴展性。所以在物聯(lián)網(wǎng)中，Influxdb數(shù)據(jù)庫一直被廣泛應(yīng)用。

想要對接時序數(shù)據(jù)庫influxDB，首先需要在EG8200邊緣計算網(wǎng)關(guān)插入SD卡，會自動安裝influxDB數(shù)據(jù)庫，并自動存儲在SD卡內(nèi)。如果未插入SD卡，則無法創(chuàng)建本地influxDB數(shù)據(jù)庫，讀寫時序數(shù)據(jù)庫節(jié)點自然也就無法使用。但是可以和遠(yuǎn)端的influxDB數(shù)據(jù)庫通信，例如局域網(wǎng)內(nèi)有influxDB數(shù)據(jù)庫。

了解了以上所有概念后，我們進入今天的主題：如何利用EG8200邊緣計算網(wǎng)關(guān)采集Modbus TCP數(shù)據(jù)存入Influxdb數(shù)據(jù)庫。

實操步驟

1. 在開始實操之前，我們看一下本次實操的目的

利用EG8200每10秒采集一次Modbus TCP數(shù)據(jù)存入Influxdb數(shù)據(jù)庫，并且每分鐘從Influxdb數(shù)據(jù)庫中取6條數(shù)據(jù)打包上報。

Modbus TCP數(shù)據(jù)采集此處不做過多贅述，下文默認(rèn)已經(jīng)采集到Modbus TCP數(shù)據(jù)。

2. 需求分析

第一步：配置并連接數(shù)據(jù)庫

第二步：創(chuàng)建一個數(shù)據(jù)庫

第三步：指定使用新創(chuàng)建的數(shù)據(jù)庫

第四步：讀取Modbus TCP數(shù)據(jù)并存入指定表中

第五步：從數(shù)據(jù)庫中讀取6個數(shù)據(jù)

第六步：將讀取出來的數(shù)據(jù)打包

第七步：將打包的數(shù)據(jù)發(fā)送到MQTT

數(shù)據(jù)上報格式：

[{
	"time": "2023-12-20 14:50:13",
	"value1": 12573,
	"value2": 12568,
	"value3": 12561,
	"value4": 12557,
	"value5": 12552,
	"value6": 12545,
	"value7": 12540,
	"value8": 12535
}, {
	"time": "2023-12-20 14:50:13",
	"value1": 12573,
	"value2": 12568,
	"value3": 12561,
	"value4": 12557,
	"value5": 12552,
	"value6": 12545,
	"value7": 12540,
	"value8": 12535
}, {
	"time": "2023-12-20 14:50:13",
	"value1": 12574,
	"value2": 12569,
	"value3": 12562,
	"value4": 12558,
	"value5": 12553,
	"value6": 12546,
	"value7": 12541,
	"value8": 12536
}, {
	"time": "2023-12-20 14:50:14",
	"value1": 12574,
	"value2": 12569,
	"value3": 12562,
	"value4": 12558,
	"value5": 12553,
	"value6": 12546,
	"value7": 12541,
	"value8": 12536
}, {
	"time": "2023-12-20 14:50:14",
	"value1": 12575,
	"value2": 12570,
	"value3": 12563,
	"value4": 12559,
	"value5": 12554,
	"value6": 12547,
	"value7": 12542,
	"value8": 12537
}, {
	"time": "2023-12-20 14:50:14",
	"value1": 12575,
	"value2": 12570,
	"value3": 12563,
	"value4": 12559,
	"value5": 12554,
	"value6": 12547,
	"value7": 12542,
	"value8": 12537
}]

3. 實操過程

第一步：配置并連接數(shù)據(jù)庫

首次連接EG8200的Influxdb數(shù)據(jù)庫按照默認(rèn)的配置即可無需修改：

第二步：創(chuàng)建一個數(shù)據(jù)庫

創(chuàng)建一個數(shù)據(jù)庫需要使用到sql語句 CREATE DATABASE myDatabase ，這條語句的意思是創(chuàng)建一個名字叫做myDatabase的數(shù)據(jù)庫。按照下圖的方式創(chuàng)建，日志窗口打印[empty]，說明數(shù)據(jù)庫創(chuàng)建成功：

第三步：指定使用新創(chuàng)建的數(shù)據(jù)庫

在第二步創(chuàng)建了一個名字叫做myDatabase的數(shù)據(jù)庫，現(xiàn)在需要使用這個數(shù)據(jù)庫，將Influxdb的配置項Database設(shè)置為myDatabase：

第四步：讀取Modbus TCP數(shù)據(jù)并存入指定表中

Modbus TCP數(shù)據(jù)采集此處不做過多贅述，這里已經(jīng)配置好Modbus TCP。

從節(jié)點庫拖出一個寫時序數(shù)據(jù)庫節(jié)點，選擇前面步驟已經(jīng)添加好的的配置信息，并將表名填寫在Measurement，這里使用myTable作為表名：

第五步：從數(shù)據(jù)庫中讀取6個數(shù)據(jù)

通過函數(shù)節(jié)點構(gòu)造sql讀取指令，通過時間查詢最近6條數(shù)據(jù)：

上圖中使用到的函數(shù)代碼：

if (global.get("startTime") == undefined || global.get("startTime") == null) global.set("startTime", new Date().getTime())
var startTime = global.get("startTime")
var temp = new Date(startTime).toISOString()
var table = "myTable"
msg.query = "SELECT * FROM" + " \"" + table + "\" " + " WHERE time > '" + temp + "' LIMIT 6"
return msg;

第六步：將讀取出來的數(shù)據(jù)打包

使用函數(shù)節(jié)點將數(shù)據(jù)封裝成對應(yīng)的json字符串：

上圖中使用到的函數(shù)代碼：

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;
}
// const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())
var dataList = msg.payload
if (dataList.length == 0) {
    node.error("數(shù)據(jù)庫無數(shù)據(jù)請檢查")
    return
}
dataList.forEach(item => item.time = dateFormat("YYYY-mm-dd HH:MM:SS", new Date(item.time)))
global.set("startTime", new Date(dataList[dataList.length - 1].time).getTime() + 1000)
msg.payload = JSON.stringify(dataList)
return msg

第七步：將打包的數(shù)據(jù)發(fā)送到MQTT

拖入一個MQTT發(fā)布節(jié)點，MQTT節(jié)點的配置（參考這里）。 

注意：MQTT配置節(jié)點導(dǎo)出時，不會攜帶用戶名和密碼。因此導(dǎo)入時需要重新錄入用戶名和密碼

客戶端ID：網(wǎng)關(guān)SN

用戶名：網(wǎng)關(guān)SN

密碼：EG87654321

發(fā)布主題：${sn}/send