项目背景
最近接到一个比较简单的任务，需求如下： 1、从MQTT服务器订阅断电报警信息然后入库到SQLServer或者MySQL数据库中 2、从MQTT服务器订阅到站点报警（0断电，1来电）、GPS信息（经纬度）、设备信号，然后在内存中缓存每个站点的这三种信息，再加上最新通信时间（接收到订阅的消息的最新时间）， 3、针对每个站点（SS打头的编码）和ClientID（设备编码），做一个HTTP GET请求接口，前端可以根据站点编码和设备编码请求该站点的数据，主要是为后期做站点在线、离线状态判断、断电告警来服务的。 程序简单的思维导图如下图所示：
本来打算使用C 写的，考虑到C 写HTTP接口相对比较麻烦，还是采用Nodejs写比较方便，因为Nodejs对于MQTT、HTTP的支持比较友好，比较适合写这种简单的后台程序。 程序大概的流程是： 1、从MQTT服务器上订阅如下的三种主题消息： 订阅主题 (1). 报警, 0断电, 1来电 /alarmSing 0
消息主题和内容示例如下： /alarmSing/865650043997457=>0
(2). GPS信息 /lbsLocation lat=022.6315208&lng=114.0741963
消息主题和内容示例如下： /lbsLocation/865650043997457=> lat=022.6315208&lng=114.0741963
(3).设备信号 /csq 18 消息主题和内容示例如下: /csq/865650043997457=>27
需要在config.yaml文件中配置好MQTT服务器的配置信息，示例如下：
代码语言：javascript复制rxmqtt:
host: 127.0.0.1
port: 8099
user: poweralarm
pwd: "poweralarm@123"
id: "mqweb_20200826_nodejs_alarm"
clean: true然后先连接MQTT服务器，设置订阅的主题并针对这三个主题分别写对应的回调处理函数。
2、在内存中维护一张站点信息的Map缓存数据结构，这里为了方便选择了TypeScript编写，
代码语言：javascript复制 stationInfos: Map<string, StationInfo>;其中StationInfo是一个站点信息类
3、在接收到MQTT服务器推送的报警（/alarmSing）、GPS信息（/lbsLocation）、设备信号（/csq ）这三种消息时，分别修改stationInfos这个Map缓存对象，并根据传递的DeviceId查询是否存在该站点，如果存在则更新设置对应的数据、最新通信时间、站点在线状态等。 4、编写http接口，根据站点编码集合站点信息Map缓存stationInfos返回对应的信息 5、当接收到站点断电消息时除了更新stationInfos缓存外，还需要将对应的断电报警信息入库。
数据库结构
目前数据库操作只涉及到两张表：站点和设备ID表Breakelectric以及断电报警记录表PowerCutHistory
MySQL数据表结构
代码语言：javascript复制DROP TABLE IF EXISTS `breakelectric`;
CREATE TABLE `breakelectric` (
`SStation` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '站点编码',
`DeviceId` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL COMMENT '设备Id',
`SStationName` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '站点名称'
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_bin ROW_FORMAT = Dynamic;
DROP TABLE IF EXISTS `powercuthistory`;
CREATE TABLE `powercuthistory` (
`SStation` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
`DeviceId` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL,
`SDateTime` datetime(0) NULL DEFAULT NULL,
`DevState` bit(1) NULL DEFAULT NULL
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_bin ROW_FORMAT = Dynamic;SQLServer数据表结构
代码语言：javascript复制DROP TABLE [dbo].[Breakelectric]
GO
CREATE TABLE [dbo].[Breakelectric] (
[SStation] varchar(255) NOT NULL ,
[DeviceId] varchar(100) NULL ,
[SStationName] varchar(255) NOT NULL
)
DROP TABLE [dbo].[PowerCutHistory]
GO
CREATE TABLE [dbo].[PowerCutHistory] (
[SStation] varchar(255) NULL ,
[DeviceId] varchar(100) NULL ,
[SDateTime] datetime NULL ,
[DevState] bit NULL
)几个关键的封装类
MQTT-TypeScript封装
为了简便，将MQTT客户端封装成一个类来使用，代码如下：
代码语言：javascript复制import mqtt = require('mqtt')
import moment = require('moment')
export interface MqttConnOpt extends mqtt.IClientOptions{}
export declare type OnMessageFunc = (topic: string, payload: Buffer) => void
declare class Topic {
public topic: string;
public qos: 0|1|2;
}
export class MQTT {
mqclient: mqtt.MqttClient;
brokerHost: string;
brokerPort: number;
subscribeTopics: Array<Topic>;
subscribeCallbacks: Map<string, OnMessageFunc>;
connOpt: MqttConnOpt;
/**
* 是否成功连接到MQTT broker
*/
connected: boolean;
constructor(host?: string | any, port?: number) {
this.brokerHost = host;
this.brokerPort = port;
this.subscribeTopics = new Array<Topic>();
this.subscribeCallbacks = new Map<string, OnMessageFunc>();
this.connected = false;
}
/**
* 订阅主题
*/
public subscribe(topic: string, qos: 0|1|2) {
this.subscribeTopics.push({topic: topic, qos: qos});
if (this.is_connected()){
this.mqclient.subscribe(topic, {qos: qos});
}
}
/**
* 设置消息数据回调函数
*/
public set_message_callback(topicPatten: string, cb: OnMessageFunc) {
this.subscribeCallbacks.set(topicPatten, cb);
}
/**
* 是否已连接到服务器
*/
public is_connected() {
// return this.mqclient.connected == true;
return this.connected == true;
}
/**
* 连接到服务器
*/
public connect(opts?: MqttConnOpt){
// 打开重新订阅
opts.resubscribe = false;
this.connOpt = opts;
this.mqclient = mqtt.connect(`mqtt://${this.brokerHost}:${this.brokerPort}`, opts);
this.mqclient.on('connect', (connack)=>{
console.log(`成功连接到服务器[${this.brokerHost}:${this.brokerPort}]`);
this.connected = true;
for (let index = 0; index < this.subscribeTopics.length; index ) {
const element = this.subscribeTopics[index];
this.mqclient.subscribe(element.topic, {qos: element.qos});
}
});
this.mqclient.on('message', (topic: string, payload: Buffer)=>{
console.log(`[${moment().format('YY-MM-DD HH:mm')}] ${this.brokerHost} ${topic}`)
this.mqclient;
this.subscribeCallbacks.forEach((val, key)=>{
if (topic.indexOf(key) != -1){
val(topic, payload);
}
});
});
this.mqclient.on('reconnect', ()=>{
console.log("重新连接")
});
this.mqclient.on('error', (err: Error)=>{
console.log(err)
});
}
/**
* 推送数据
*/
public publish(topic: string, message: string, qos: 0|1|2) {
this.mqclient.publish(topic, message, {qos: qos, retain: false})
}
}其中，需要注意的一点就是MQTT服务器有可能意外重启或者其他原因断开，这时需要断线重连。在C 、C#、Java等语言中可以开启一个断线重连监测线程，每隔一段时间监测与MQTT服务器的连接情况，如果断线则重新连接。
yaml文件配置类对象
为了方便这里采用yaml文件作为配置文件，之前使用C 时也常用xml、ini、yaml作为配置文件，Java SpringBoot也常用yml或yaml作为配置文件。 我的yaml配置文件如下图所示：
代码语言：javascript复制rxmqtt:
host: 127.0.0.1
port: 8099
user: poweralarm
pwd: "poweralarm@123"
id: "mqweb_20200826_nodejs_alarm"
clean: true
# dbsql:
# host: 127.0.0.1
# port: 1433
# user: sa
# pwd: "123456"
# database: EMCHNVideoMonitor
dbsql:
host: 127.0.0.1
port: 3306
user: root
pwd: "123456"
database: EMCHNVideoMonitor
redis:
host: 127.0.0.1
port: 7001
pwd: 123456
index: 3
http: 3000
rpcUrl: 127.0.0.1:18885
enableMqtt: true
enableDB: true
enableRedis: true
enableWS: true
enableRPC: true
offlineTimeout: 90000
cacheInterval: 10000针对上面的yaml配置文件，编写对应的yaml配置读取类，如下所示：
代码语言：javascript复制import YAML = require('yaml')
import fs = require('fs')
declare interface MqttConnOpt{
host: string;
port: number;
user: string;
pwd: string;
clean: boolean;
id: string;
}
declare interface DBConnOpt{
host: string;
port: number;
user: string;
pwd: string;
database: string;
}
declare interface RedisConnOpt{
host: string;
port: number;
pwd: string;
db: number;
}
export {
MqttConnOpt,
DBConnOpt,
RedisConnOpt,
Config,
}
class Config {
rxmqtt: MqttConnOpt;
dbsql: DBConnOpt;
redis: RedisConnOpt;
/**
* http 端口
*/
http: number;
/**
* rpcUrl 服务器地址
*/
rpcUrl: string;
/**
* 是否启用mqtt
*/
enableMqtt: boolean;
/**
* 是否启用sqlServer或者mysql数据库
*/
enableDB: boolean;
/**
* 是否启用redis
*/
enableRedis: boolean;
/**
* 是否启用websocket
*/
enableWS: boolean;
/**
* 是否启用RPC
*/
enableRPC: boolean;
/**
* 离线超时时间, 毫秒
*/
offlineTimeout: number;
/**
* 缓存存储间隔, 毫秒
*/
cacheInterval: number;
constructor(){
try{
let buffer = fs.readFileSync('config.yaml', 'utf8');
let config = YAML.parse(buffer);
this.rxmqtt = config['rxmqtt'];
this.dbsql = config['dbsql'];
this.redis = config['redis'];
this.http = config['http'];
this.rpcUrl = config['rpcUrl'];
this.enableMqtt = config['enableMqtt'];
this.enableDB = config['enableDB'];
this.enableRedis = config['enableRedis'];
this.enableWS = config['enableWS'];
this.enableRPC = config['enableRPC'];
this.offlineTimeout = config['offlineTimeout'];
this.cacheInterval = config['cacheInterval'];
}catch(err){
console.log(err)
}
}
/**
* save
*/
public save() {
try{
fs.writeFileSync('config.yaml', YAML.stringify(this))
}catch(err){
console.log(err)
}
}
}其实使用yaml这个第三方库结合typescript读写yaml文件还是比较方便的。
数据操作类的封装
mysql操作类
nodejs中可以使用mariadb或者sequelize等库操作mysql数据库，这里使用mariadb这个库 MariaDBClient.ts
代码语言：javascript复制import mariadb = require('mariadb')
import { StationInfo } from './StationInfo'
import moment = require('moment')
// 定义数据查询回调接口
export declare type OnQueryInfoReqCallback = (err: Error, rc: Array<any>) => void
// 定义入库回调接口
export declare type OnRecordReqCallback = (err: Error, rc: boolean) => void
export class MariaDBClient {
dbpool: mariadb.Pool;
host: string;
port: number;
user: string;
password: string;
dbName: string;
connected: boolean;
// 站点信息 Map
public stationInfos: Map<string, StationInfo>;
constructor(username: string, password: string, dbName: string, host?: string | any, port?: number) {
this.host = host;
this.port = port;
this.user = username;
this.password = password;
this.dbName = dbName;
this.connected = false;
this.stationInfos = new Map<string, StationInfo>();
// 初始化mariadb数据库客户端
this.initMariadb();
// 加载站点信息到内存中
this.getStationInfo();
}
/**
* 初始化mariadb数据库客户端
*/
public initMariadb() {
this.dbpool = mariadb.createPool({
host: this.host,
port: this.port,
user: this.user,
password: this.password,
database: this.dbName,
connectionLimit: 10,
});
}
/**
* 是否已连接到MariaDB数据库
*/
public is_connected() {
return this.connected == true;
}
/**
* 获取站点信息
*/
public async getStationInfo() {
let conn;
try {
conn = await this.dbpool.getConnection();
const rows = await conn.query("SELECT SStation, DeviceId, SStationName from Breakelectric WHERE SStation != '' AND DeviceId != '';");
for (let i = 0; i < rows.length; i ) {
const it = rows[i];
const SStation = it['SStation'];
this.stationInfos.has
if (!this.stationInfos.has(SStation)) {
let si = new StationInfo();
si.SStation = it['SStation'];
si.DeviceId = it['DeviceId'];
si.SStationName = it['SStationName'];
console.log(`第${i 1}个站点：站点编码：${si.SStation},设备Id: ${si.DeviceId},站点名称：${si.SStationName}`);
this.stationInfos.set(SStation, si);
}
}
} catch (e) {
console.error(e);
} finally {
if (conn) conn.release(); //release to pool
}
}
/**
*
* @param record 获取站点列表
*/
public async getStationList(cb: OnQueryInfoReqCallback) {
let conn;
try {
conn = await this.dbpool.getConnection();
const rows = await conn.query("SELECT SStation, DeviceId, SStationName from Breakelectric WHERE SStation != '' AND DeviceId != '';");
let stationList = new Array<any>();
for (let i = 0; i < rows.length; i ) {
const rowItem = rows[i];
let iitem = {
'SStation': rowItem['SStation'],
'DeviceId': rowItem['DeviceId'],
'SStationName': rowItem['SStationName']
}
stationList.push(iitem);
}
if (cb) cb(null, stationList);
} catch (e) {
console.error(e);
if (cb) cb(e, null);
} finally {
if (conn) conn.release(); //release to pool
}
}
/**
* 增、删、改、查 CRUD操作 API
*/
/**
*
* @param record 插入断电报警记录
* @param cb
*/
public async insertStationRecord(record: any) {
if (record === null) {
return;
}
let sql1 = "INSERT INTO `emchnvideomonitor`.`powercuthistory` (`SStation`, `DeviceId`, `SDateTime`, `DevState`) VALUES";
let conn: mariadb.PoolConnection;
try {
conn = await this.dbpool.getConnection();
var sqlstr = sql1;
let SStation = record.SStation; // 站点名称
let DeviceId = record.DeviceId; // 设备Id
let SDateTime = record.SDateTime; // 时间
let DevState = record.DevState; // 状态（0停电，1来电）
var it = `('${SStation}','${DeviceId}','${SDateTime}',${DevState})`;
sqlstr = it;
console.log(sqlstr);
await conn.query(sqlstr);
// if (cb) cb(null, true);
} catch (e) {
console.error('插入断电报警信息失败,',e);
// if (cb) cb(e, false);
} finally {
if (conn) conn.release(); //release to pool
}
}
}sqlserver操作类
nodejs中可以使用tedious、mmsql、sequelize等库操作sqlserver数据库，这里采用mssql封装sqlserver操作： MariaDBClient.ts
代码语言：javascript复制import mssql = require('mssql');
// 定义数据查询回调接口
export declare type OnQueryCallback = (err: Error, rc: any) => void
export declare type OnExecCallback = (err: Error, rc: boolean) => void
export class MSSQLDBClient {
// 数据库连接字符串
// 连接方式："mssql://用户名:密码@ip地址:1433(默认端口号)/数据库名称"
constr: string;
constructor(username: string, password: string, host: string, port: number, dbName: string) {
this.constr = `mssql://${username}:${password}@${host}:${port}/${dbName}`;
mssql.connect(this.constr).then(function () {
console.log('----------------');
console.log('-数据库登录成功-');
console.log('----------------');
}).catch(function (err) {
console.log(err);
})
}
/**
* 根据sql脚本查询数据库中的表
* @param strSql SQL脚本
* @param cb 查询结果的回调函数
*/
public async query(strSql: string, cb: OnQueryCallback) {
try {
await mssql.connect(this.constr).then(function() {
new mssql.Request().query(strSql).then(function(result) {
// console.log(result);
if (cb) cb(null, result);
}).catch(function(err) {
console.log(err);
if (cb) cb(err, null);
});
// Stored Procedure
}).catch(function(err) {
console.log(err);
if (cb) cb(err, null);
})
} catch (err) {
console.log(err);
if (cb) cb(err, null);
}
}
/**
*
* @param strSql SQL脚本
* @param cb 执行SQL脚本的回调
*/
public async exec(strSql: string, cb: OnExecCallback) {
await mssql.connect(this.constr, function () {
mssql.query(strSql, function (err, data) {
if (err) {
if (cb) cb(err, false);
} else {
if (cb) cb(null, true);
mssql.close();
}
});
}
);
}
}主服务类 service.ts
代码语言：javascript复制import moment = require('moment')
import sql = require('mssql')
import { Config } from './config'
import { MQTT } from './mq'
import * as http from 'http'
import { StationInfo } from './StationInfo'
// const mssqlDBClient = require('./db');
// import { MSSQLDBClient } from './MSSQLDBClient'
import { MariaDBClient } from './MariaDBClient'
/**********************************************************************************************************
* 1、从MQTT服务器订阅断电报警信息然后入库到SQLServer数据库中
* 2、从MQTT服务器订阅到站点报警（0断电，1来电）、GPS信息、设备信号，然后在内存中分别缓存每个站点的这三种信息，再加上最新通信时间（接收到订阅的消息的最新时间），
* 然后针对每个站点（SS打头的编码）和ClientID（设备编码），做一个HTTP GET请求接口，前端可以根据站点编码和设备编码请求该站点的数据，
* 主要是为后期做站点在线、离线来服务的。
*/
export class Service {
mqttList: Array<MQTT>; // mqtt客户端列表
stationInfos: Map<string, StationInfo>;
config: Config;
Server: http.Server;
App: any;
// mssqlDBClient: MSSQLDBClient;
mySQLDBClient: MariaDBClient;
// 构造函数
constructor(app:any, server:http.Server) {
this.Server = server;
this.App = app;
this.config = new Config();
// 初始化配置
// this.mssqlDBClient = new MSSQLDBClient(
// this.config.dbsql.user,
// this.config.dbsql.pwd,
// this.config.dbsql.host,
// this.config.dbsql.port,
// this.config.dbsql.database
// );
// 创建数据库客户端
this.mySQLDBClient = new MariaDBClient(
this.config.dbsql.user,
this.config.dbsql.pwd,
this.config.dbsql.database,
this.config.dbsql.host,
this.config.dbsql.port
);
this.mqttList = new Array<MQTT>();
this.stationInfos = new Map<string, StationInfo>();
// 建立客户端连接
if (this.config.enableMqtt) {
this.connectMqtt();
}
// 加载缓存数据到内存中
this.LoadStations();
// 定时检查站点是否在线
// this.taskCheckStationOnline();
// 定时存储站点数据缓存
this.taskStoreStationData();
// 定时重载站点信息
this.timerLoadStationInfo();
// 初始化http请求
this.initApp();
}
/**
* 定时加载站点信息
*/
public timerLoadStationInfo() {
setInterval(async ()=>{
// this.getStationInfo();
await this.mySQLDBClient.getStationInfo();
this.stationInfos = this.mySQLDBClient.stationInfos;
}, 120*1000);
}
/**
* 加载缓存数据到内存中
*/
public async LoadStations() {
// 加载最后的缓存数据
// await this.LoadRedisData();
// 加载站点信息
// await this.getStationInfo();
await this.mySQLDBClient.getStationInfo();
this.stationInfos = this.mySQLDBClient.stationInfos;
}
/**
* 定时检查站点的状态
*/
public taskCheckStationOnline() {
setInterval(()=>{
this.timerCheckOnline();
}, this.config.offlineTimeout);
}
/**
* 定时存储站点缓存数据
*/
public taskStoreStationData() {
setInterval(()=>{
// this.timerStorStationData();
// this.taskStorNewData();
}, this.config.cacheInterval);
}
/**
* 检查站点是否在线
*/
public timerCheckOnline() {
let stcodeIds = [];
this.stationInfos.forEach((val, key) => {
let previous_online = val.Online;
val.checkOnline();
// 如果先前在线，现在离线
if (previous_online && !val.Online) {
stcodeIds.push(val.SStation);
// this.sendHeart2WSClient(val.toWebHeartString());
}
})
}
/**
* 初始化http请求
*/
public initApp() {
if (!this.App) {
return;
}
// 路由接口
// 获取所有的站点编码和设备ID映射列表
this.App.get('/api/getStationList', (req, res) => {
let stationList = [];
this.stationInfos.forEach((val, key) => {
stationList.push({
'SStation': val.SStation,
'DeviceId': val.DeviceId,
'SStationName': val.SStationName
});
});
res.send({
rc: true,
data: stationList
})
});
// 根据站点编码获取当前站点信息
this.App.get('/api/getAlarmInfo/:stcode', (req, res) => {
let { stcode } = req.params;
if (this.stationInfos.has(stcode)) {
let item = this.stationInfos.get(stcode);
return res.send({
rc: true,
data: item
})
} else {
res.send({
rc: false,
data: '站点编码不存在'
})
}
})
// 获取所有站点断电报警、设备信号、经纬度等信息
this.App.get('/api/getAllStationInfos', (req, res) => {
let stationList = [];
this.stationInfos.forEach((val, key) => {
stationList.push(val);
})
res.send({
rc: true,
data: stationList
})
})
}
/**
* 获取站点信息
*/
public async getStationInfo() {
// 一、SQLServer
// let strSql = "SELECT SStation, DeviceId, SStationName from Breakelectric WHERE SStation != '' AND DeviceId != '';";
// this.mssqlDBClient.query(strSql, (err, result) => {
// if (result == null || result == '' || result.recordsets[0] == undefined
// || result.recordsets[0] == null) {
// return;
// }
// let resultArray = result.recordsets[0];
// if (resultArray != null && resultArray.length > 0) {
// for (let i = 0; i < resultArray.length; i ) {
// // this.stationList.push(resultArray[i]);
// let iitem = resultArray[i];
// console.log(`第${i 1}个站点，SStation：${iitem.SStation},DeviceId: ${iitem.DeviceId},SStationName: ${iitem.SStationName}`);
// // console.log(resultArray[i]);
// let stcode = iitem['SStation'];
// if (!this.stationInfos.has(stcode)) {
// this.stationInfos.set(stcode, new StationInfo());
// }
// var si = this.stationInfos.get(stcode);
// si.SStation = iitem['SStation'];
// si.DeviceId = iitem['DeviceId'];
// si.SStationName = iitem['SStationName'];
// }
// console.log(JSON.stringify(this.stationInfos));
// }
// });
// 二、MariaDB
this.mySQLDBClient.getStationList((err, result) => {
if (!err && result != null && result != []) {
for (let i = 0; i < result.length; i ) {
let iitem = result[i];
let SStation = iitem['SStation'];
if (!this.stationInfos.has(SStation)) {
this.stationInfos.set(SStation, new StationInfo());
}
var si = this.stationInfos.get(SStation);
si.SStation = iitem['SStation'];
si.DeviceId = iitem['DeviceId'];
si.SStationName = iitem['SStationName'];
console.log(`第${i 1}个站点：站点编码：${si.SStation},设备Id: ${si.DeviceId},站点名称：${si.SStationName}`);
}
}
})
}
/**
* 连接MQTT服务器
*/
public connectMqtt() {
var it = new MQTT(this.config.rxmqtt.host, this.config.rxmqtt.port);
// 订阅主题
// 1. 报警, 0断电, 1来电
// /alarmSing 0
it.subscribe('/alarmSing', 0);
// 2. GPS信息
// /lbsLocation lat=022.6315208&lng=114.0741963
it.subscribe('/lbsLocation', 0);
// 3.设备信号
// /csq 18
it.subscribe('/csq', 0);
it.set_message_callback('/alarmSing', this.handleAlarmSing.bind(this));
it.set_message_callback('/lbsLocation', this.handleGpsLocation.bind(this));
it.set_message_callback('/csq', this.handleCsq.bind(this));
it.connect({
username: this.config.rxmqtt.user,
password: this.config.rxmqtt.pwd,
clientId: this.config.rxmqtt.id,
clean: this.config.rxmqtt.clean,
});
this.mqttList.push(it);
}
/**
* 断电报警数据处理函数
*/
handleAlarmSing(topic: string, payload: Buffer) {
console.log(`断电报警数据: ${topic}=>${payload.toString()}`);
const topics = topic.split('/');
// /alarmSing/867814045313299=>1
console.log('DeviceId', topics[1]);
if (topics[1] == 'alarmSing') {
let deviceId = topics[2];
console.log('设备Id: ', deviceId);
let alarmDevState = parseInt(payload.toString());
console.log('断电报警DevState: ', alarmDevState == 0 ? '停电' : '来电');
// 根据DeviceId查询对应的站点编码SStation
let stcode = '';
this.stationInfos.forEach((val, key) => {
if (val.DeviceId == deviceId) {
stcode = key;
// 更新该站点的通信时间以及断电报警信息
let comTime = moment().format('YYYY-MM-DD HH:mm:ss');
var si = this.stationInfos.get(stcode);
let strStcode = si.SStation;
si.alarmSing = alarmDevState;
si.CommTime = comTime;
si.Online = true;
this.stationInfos.set(stcode, si);
// 将断电报警信息做入库处理
// this.powerCutAlarmStore({
// SStation: strStcode,
// DeviceId: deviceId,
// SDateTime: comTime,
// DevState: alarmDevState
// });
this.mySQLDBClient.insertStationRecord({
SStation: strStcode,
DeviceId: deviceId,
SDateTime: comTime,
DevState: alarmDevState
})
}
});
}
}
/**
* GPS信息数据处理函数
*/
handleGpsLocation (topic: string, payload: Buffer) {
console.log(`GPS信息数据: ${topic}=>${payload.toString()}`);
const topics = topic.split('/');
// /lbsLocation/867814045313299=>lat=022.7219409&lng=114.0222168
if (topics[1] == 'lbsLocation') {
let deviceId = topics[2];
console.log('设备Id: ', deviceId);
let strPayload = payload.toString();
let strLatitude = strPayload.substring(strPayload.indexOf("lat=") 4, strPayload.indexOf("&"));
let latitude = parseFloat(strLatitude);
console.log('latitude: ', latitude);
let strLongitude = strPayload.substring(strPayload.indexOf("lng=") 4);
let longitude = parseFloat(strLongitude.toString());
console.log('longitude: ', longitude);
// 根据DeviceId查询对应的站点编码SStation
let stcode = '';
this.stationInfos.forEach((val, key) => {
if (val.DeviceId == deviceId) {
stcode = key;
// 更新该站点的通信时间以及经纬度信息
let comTime = moment().format('YYYY-MM-DD HH:mm:ss');
var si = this.stationInfos.get(stcode);
si.Online = true;
si.longitude = longitude;
si.latitude = latitude;
si.CommTime = comTime;
this.stationInfos.set(stcode, si);
}
});
}
}
/**
* 设备信号数据处理函数
*/
handleCsq(topic: string, payload: Buffer) {
console.log(`设备信号数据: ${topic}=>${payload.toString()}`);
const topics = topic.split('/');
// /csq/867814045454838=>20
if (topics[1] == 'csq') {
let deviceId = topics[2];
console.log('设备Id: ', deviceId);
let csq = parseInt(payload.toString());
console.log('设备信号: ', csq);
// 根据DeviceId查询对应的站点编码SStation
let stcode = '';
this.stationInfos.forEach((val, key) => {
if (val.DeviceId == deviceId) {
stcode = key;
// 更新该站点的通信时间以及csq信号值
let comTime = moment().format('YYYY-MM-DD HH:mm:ss');
var si = this.stationInfos.get(stcode);
si.Online = true;
si.csq = csq;
si.CommTime = comTime;
this.stationInfos.set(stcode, si);
}
});
}
}
/**
* 站点断电报警数据存储
*/
public async powerCutAlarmStore(alarmRecord: any) {
var SStation = alarmRecord.SStation;
var DeviceId = alarmRecord.DeviceId;
var SDateTime = alarmRecord.SDateTime;
var DevState = alarmRecord.DevState;
let strInsert = "INSERT INTO powercuthistory(SStation, DeviceId, SDateTime, DevState) VALUES";
strInsert = `('${SStation}','${DeviceId}','${SDateTime}',${DevState})`;
// this.mssqlDBClient.exec(strInsert, (err, rc) => {
// if (err) {
// console.log('插入报警数据出错:', err);
// }
// })
}
}app.js
这里为了简便，我直接使用express生成器生成了项目的基本框架，对应的app.js文件如下：
代码语言：javascript复制var createError = require('http-errors');
var express = require('express');
var app = express();
var path = require('path');
var logger = require('morgan');
// var indexRouter = require('./routes/index');
// var usersRouter = require('./routes/users');
var app = express();
// view engine setup
app.set('views', path.join(__dirname, 'views'));
app.set('view engine', 'jade');
app.use(logger('dev'));
app.use(express.json());
app.use(express.urlencoded({ extended: false }));
app.use(express.static(path.join(__dirname, 'public')));
// app.use('/', indexRouter);
// app.use('/users', usersRouter);
module.exports = app;bin/www
在bin/www文件中创建了service类的实例，然后读取config配置，并启动相关服务。注意：这里需要将app和server传入到service对象中，在service对象中编写http接口，这样就能保证http接口和站点信息缓存共享同一份数据了，如果将http接口写在app.js或者routes/api.js中，创建两个service对象，就不能保证站点信息缓存信息的数据同步了。
代码语言：javascript复制#!/usr/bin/env node
/**
* Module dependencies.
*/
var app = require('../app');
var debug = require('debug')('hnmqalarmstore:server');
var http = require('http');
var config_1 = require('../config');
var Service_1 = require('../service');
var config = new config_1.Config();
/**
* Get port from environment and store in Express.
*/
var port = normalizePort(process.env.PORT || config.http);
app.set('port', port);
/**
* Create HTTP server.
*/
var server = http.createServer(app);
// 服务对象
new Service_1.Service(app, server);
// catch 404 and forward to error handler
app.use(function(req, res, next) {
next(createError(404));
});
// error handler
app.use(function(err, req, res, next) {
// set locals, only providing error in development
res.locals.message = err.message;
res.locals.error = req.app.get('env') === 'development' ? err : {};
// render the error page
res.status(err.status || 500);
res.render('error');
});
/**
* Listen on provided port, on all network interfaces.
*/
server.listen(port);
server.on('error', onError);
server.on('listening', onListening);
/**
* Normalize a port into a number, string, or false.
*/
function normalizePort(val) {
var port = parseInt(val, 10);
if (isNaN(port)) {
// named pipe
return val;
}
if (port >= 0) {
// port number
return port;
}
return false;
}
/**
* Event listener for HTTP server "error" event.
*/
function onError(error) {
if (error.syscall !== 'listen') {
throw error;
}
var bind = typeof port === 'string'
? 'Pipe ' port
: 'Port ' port;
// handle specific listen errors with friendly messages
switch (error.code) {
case 'EACCES':
console.error(bind ' requires elevated privileges');
process.exit(1);
break;
case 'EADDRINUSE':
console.error(bind ' is already in use');
process.exit(1);
break;
default:
throw error;
}
}
/**
* Event listener for HTTP server "listening" event.
*/
function onListening() {
var addr = server.address();
var bind = typeof addr === 'string'
? 'pipe ' addr
: 'port ' addr.port;
debug('Listening on ' bind);
}使用到的一些第三方库
yaml、mssql、mariadb、mqtt、express等，对应的项目的package.json文件如下：
代码语言：javascript复制“name”: “hnmqalarmstore”,
“version”: “0.0.0”,
“private”: true,
“scripts”: {
“start”: “node ./bin/www”
},
“dependencies”: {
“@js-joda/core”: “^3.0.0”,
“body-parser”: “^1.19.0”,
“cookie-parser”: “^1.4.5”,
“debug”: “~2.6.9”,
“express”: “^4.16.4”,
“express-session”: “^1.17.1”,
“http-errors”: “^1.8.0”,
“jade”: “^1.11.0”,
“mariadb”: “^2.4.2”,
“moment”: “^2.27.0”,
“morgan”: “^1.9.1”,
“mqtt”: “^4.2.1”,
“mssql”: “^6.2.1”,
“yaml”: “^1.10.0”
},
“devDependencies”: {
“nodemon”: “^2.0.4”
}
}
