如何利用EG8200边缘计算网关采集Modbus TCP数据存入Influxdb数据库

Influxdb数据库是什么？边缘计算网关如何对接Influxdb数据库？

Influxdb是非关系型数据库中的时序数据库，全称时间序列数据库数据库。MySQL作为一种传统的关系型数据库，其扩展性相对有限，特别是在处理大规模并发读写时，其性能可能会受到较大的影响。而时序数据库则通过分布式架构设计，能够更好地支持大规模并发读写，具有更好的扩展性。所以在物联网中，Influxdb数据库一直被广泛应用。

想要对接时序数据库influxDB，首先需要在EG8200边缘计算网关插入SD卡，会自动安装influxDB数据库，并自动存储在SD卡内。如果未插入SD卡，则无法创建本地influxDB数据库，读写时序数据库节点自然也就无法使用。但是可以和远端的influxDB数据库通信，例如局域网内有influxDB数据库。

了解了以上所有概念后，我们进入今天的主题：如何利用EG8200边缘计算网关采集Modbus TCP数据存入Influxdb数据库。

实操步骤

1. 在开始实操之前，我们看一下本次实操的目的

利用EG8200每10秒采集一次Modbus TCP数据存入Influxdb数据库，并且每分钟从Influxdb数据库中取6条数据打包上报。

Modbus TCP数据采集此处不做过多赘述，下文默认已经采集到Modbus TCP数据。

2. 需求分析

第一步：配置并连接数据库

第二步：创建一个数据库

第三步：指定使用新创建的数据库

第四步：读取Modbus TCP数据并存入指定表中

第五步：从数据库中读取6个数据

第六步：将读取出来的数据打包

第七步：将打包的数据发送到MQTT

数据上报格式：

[{
	"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. 实操过程

第一步：配置并连接数据库

首次连接EG8200的Influxdb数据库按照默认的配置即可无需修改：

第二步：创建一个数据库

创建一个数据库需要使用到sql语句 CREATE DATABASE myDatabase ，这条语句的意思是创建一个名字叫做myDatabase的数据库。按照下图的方式创建，日志窗口打印[empty]，说明数据库创建成功：

第三步：指定使用新创建的数据库

在第二步创建了一个名字叫做myDatabase的数据库，现在需要使用这个数据库，将Influxdb的配置项Database设置为myDatabase：

第四步：读取Modbus TCP数据并存入指定表中

Modbus TCP数据采集此处不做过多赘述，这里已经配置好Modbus TCP。

从节点库拖出一个写时序数据库节点，选择前面步骤已经添加好的的配置信息，并将表名填写在Measurement，这里使用myTable作为表名：

第五步：从数据库中读取6个数据

通过函数节点构造sql读取指令，通过时间查询最近6条数据：

上图中使用到的函数代码：

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;

第六步：将读取出来的数据打包

使用函数节点将数据封装成对应的json字符串：

上图中使用到的函数代码：

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;
}
// const fmt = dateFormat("YYYY-mm-dd HH:MM:SS", new Date())
var dataList = msg.payload
if (dataList.length == 0) {
    node.error("数据库无数据请检查")
    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

第七步：将打包的数据发送到MQTT

拖入一个MQTT发布节点，MQTT节点的配置（参考这里）。 

注意：MQTT配置节点导出时，不会携带用户名和密码。因此导入时需要重新录入用户名和密码

客户端ID：网关SN

用户名：网关SN

密码：EG87654321

发布主题：${sn}/send