设备离线连续推送

MqttServer/V3MqttHandle.go

@@ -237,6 +237,11 @@ func AsyncFuncV3(r_Device *Device.Device, Ms_project Ms2m_Project, msg []byte) {
 				}
 			}
 
+			// 报警数据没有按照两分钟间隔推送
+			if Warning_r.T_tp == 1 || Warning_r.T_tp == 2 || Warning_r.T_tp == 4 || Warning_r.T_tp == 5 {
+
+				continue
+			}
 			//tp	报警类型
 			// 1	温度超上限报警
 			// 2	温度超下限报警

TimeTask/OrmLog.go

@@ -1,5 +1,12 @@
 package TimeTask
 
+//import (
+//	"fmt"
+//	"io/ioutil"
+//	"os"
+//	"time"
+//)
+//
 //var OrmLogOrmLog *os.File
 //// ORM 日志清理
 //func OrmLog()  {

conf/app.conf

@@ -28,8 +28,8 @@ Redis_dbNum = "1"
 # Mqtt
 # 192.168.0.7
  # mqttjxit 140.246.233.197   mqttlodr 203.195.71.200  mqttyuht 203.57.71.139
-MqttServer_id = "mqttjxit"
-MqttServer_Url = "140.246.233.197:1883"
+MqttServer_id = "text_cold_mqtt"
+MqttServer_Url = "203.57.71.139:1883"
 MqttServer_ClientID = "text_cold_mqtt"
 MqttServer_Username = "admin"
 MqttServer_Password = "8f9qRNixEMhCVrF"

lib/MinuteToDataTime.go

@@ -1,7 +1,9 @@
 package lib
 
 import (
+	"fmt"
 	"strconv"
+	"time"
 )
 
 /*
@@ -56,3 +58,30 @@ func MinuteToDataTime(t int) string {
 
 	return str
 }
+
+func TimeSinceToString(startTime, endTime time.Time) string {
+	str := ""
+	// 计算时间差
+	duration := endTime.Sub(startTime)
+	// 将时间差表示为天、小时和分钟
+	days := int(duration.Hours() / 24)
+	hours := int(duration.Hours()) % 24
+	minutes := int(duration.Minutes()) % 60
+	seconds := int(duration.Seconds()) % 60
+
+	fmt.Printf("时间差为 %d 天 %d 小时 %d 分钟\n", days, hours, minutes)
+	if days > 0 {
+		str += strconv.Itoa(days) + "天 "
+	}
+	if hours > 0 {
+		str += strconv.Itoa(hours) + "小时 "
+	}
+	if minutes > 0 {
+		str += strconv.Itoa(minutes) + "分钟 "
+	}
+	if seconds > 0 {
+		str += strconv.Itoa(seconds) + "秒 "
+	}
+
+	return str
+}

main.go

@@ -1,8 +1,7 @@
 package main
 
 import (
-	"Cold_mqtt/MqttServer"
-	"Cold_mqtt/Nats"
+	"Cold_mqtt/Plugins"
 	"Cold_mqtt/TimeTask"
 	"Cold_mqtt/conf"
 	"Cold_mqtt/logs"
@@ -33,13 +32,12 @@ func main() {
 	beego.BConfig.Listen.HTTPPort = conf.HTTPPort                               //监听端口  本地：8518  线上：8528
 
 	logs.Println("MqttServer", " ======= 项目启动 ========")
-	go MqttServer.Run_MqttServer() // MQTT 通讯
-	//go MqttServer.MqttPolling() // MQTT 设备轮询
-	go MqttServer.DeviceMqttMap_go() //// 缓存数据发送-确保设备在休眠后 能收到数据
-
-	go Nats.NatsInit() // Nats 通讯
-
-	go TimeTask.Init() // 时间任务
+	//go MqttServer.Run_MqttServer() // MQTT 通讯
+	////go MqttServer.MqttPolling() // MQTT 设备轮询
+	//go MqttServer.DeviceMqttMap_go()                  //// 缓存数据发送-确保设备在休眠后 能收到数据
+	//go Nats.NatsInit()                                // Nats 通讯
+	go TimeTask.Init()                                // 时间任务
+	go Plugins.ContinuousPushOfOfflineDeviceWarning() //设备离线报警连续推送
 
 	beego.Run()
 }

models/Device/Device.go

@@ -180,3 +180,17 @@ func Update_Device_new(r Device, cols ...string) {
 
 	return
 }
+
+// 获取离线设备
+func Read_获取BX100W离线设备() []Device {
+
+	o := orm.NewOrm()
+	// 也可以直接使用 Model 结构体作为表名
+
+	qs := o.QueryTable(new(Device))
+	var r []Device
+	//qs.Filter("T_mqttid", conf.MqttServer_id).Filter("T_model", "BX100W").Filter("T_State", 1).Filter("T_online", 0).All(&r)
+	qs.Filter("T_pid", 242).Filter("T_State", 1).Exclude("T_online", 1).All(&r)
+
+	return r
+}

models/Warning/Warning.go

@@ -195,26 +195,26 @@ func Add_Warning(m Warning) (id int64, err error) {
 	}
 
 	// 插入分表
-	warningDB := WarningDB{
-		T_pid:      int32(m.T_pid),
-		T_tp:       m.T_tp,
-		T_sn:       m.T_sn,
-		T_D_name:   m.T_D_name,
-		T_id:       int32(m.T_id),
-		T_DS_name:  m.T_DS_name,
-		T_Remark:   m.T_Remark,
-		T_Ut:       m.T_Ut,
-		T_fUt:      m.T_fUt,
-		T_Text:     m.T_Text,
-		T_Log:      m.T_Log,
-		T_Msid:     m.T_Msid,
-		T_State:    m.T_State,
-		CreateTime: m.CreateTime,
-		UpdateTime: m.UpdateTime,
-	}
-	warningDB.Create("warning_" + m.T_Ut.Format("2006_01"))
+	//warningDB := WarningDB{
+	//	T_pid:      int32(m.T_pid),
+	//	T_tp:       m.T_tp,
+	//	T_sn:       m.T_sn,
+	//	T_D_name:   m.T_D_name,
+	//	T_id:       int32(m.T_id),
+	//	T_DS_name:  m.T_DS_name,
+	//	T_Remark:   m.T_Remark,
+	//	T_Ut:       m.T_Ut,
+	//	T_fUt:      m.T_fUt,
+	//	T_Text:     m.T_Text,
+	//	T_Log:      m.T_Log,
+	//	T_Msid:     m.T_Msid,
+	//	T_State:    m.T_State,
+	//	CreateTime: m.CreateTime,
+	//	UpdateTime: m.UpdateTime,
+	//}
+	//warningDB.Create("warning_" + m.T_Ut.Format("2006_01"))
 
-	//INSERT("warning_"+m.T_Ut.Format("2006_01"), m.Id)   # 旧 备份
+	INSERT("warning_"+m.T_Ut.Format("2006_01"), m.Id) // 旧 备份
 
 	return id, err
 }
@@ -266,3 +266,29 @@ func Add_DeviceLogs(T_tp int, r_Device Device.Device, T_Remark string) Warning {
 
 	return Warning_r
 }
+
+// 指定设备与类型最新的记录
+func Read_T_snT_tp(T_sn string, T_tp int) Warning {
+
+	o := orm.NewOrm()
+	// 也可以直接使用 Model 结构体作为表名
+
+	qs := o.QueryTable(new(Warning))
+	var r Warning
+	qs.Filter("T_sn", T_sn).Filter("T_tp", T_tp).OrderBy("-T_Ut").One(&r)
+
+	return r
+}
+
+// 指定设备与类型消息ID最新的记录
+func Read_T_snT_tpT_Msid(T_sn string, T_tp, T_Msid int) Warning {
+
+	o := orm.NewOrm()
+	// 也可以直接使用 Model 结构体作为表名
+
+	qs := o.QueryTable(new(Warning))
+	var r Warning
+	qs.Filter("T_sn", T_sn).Filter("T_tp", T_tp).Filter("T_Msid", T_Msid).OrderBy("-T_Ut").One(&r)
+
+	return r
+}

models/Warning/WarningBack.go

@@ -1,29 +1,40 @@
 package Warning
 
-//var Db *gorm.DB
-//var SysSeekCREATE sync.Map
-//
-//func init() {
-//	// 连接数据库
-//	var err error
-//	Db, err = gorm.Open("mysql", conf.MysqlServer_Username+":"+conf.MysqlServer_Password+"@tcp("+conf.MysqlServer_UrlPort+")/"+conf.MysqlServer_Database+"?charset=utf8mb4&loc=Local&parseTime=True")
-//	if err != nil {
-//		log.Fatal(err)
-//	}
-//	// SetMaxIdleConns 设置空闲连接池中的最大连接数。
-//	Db.DB().SetMaxIdleConns(10)
-//
-//	// SetMaxOpenConns 设置数据库连接最大打开数。
-//	Db.DB().SetMaxOpenConns(600)
-//
-//	// SetConnMaxLifetime 设置可重用连接的最长时间
-//	Db.DB().SetConnMaxLifetime(time.Hour)
-//	//Db.AutoMigrate(&DeviceWarning{})
-//	// 开启 Logger, 以展示详细的日志
-//	Db.LogMode(true)
-//	Db.SetLogger(log.New(logs.LogxOrm, "\r\nWarningBack:", 0))
-//
-//}
+import (
+	"Cold_mqtt/conf"
+	"Cold_mqtt/logs"
+	"github.com/jinzhu/gorm"
+	_ "github.com/jinzhu/gorm/dialects/mysql"
+	"log"
+	"strings"
+	"sync"
+	"time"
+)
+
+var Db *gorm.DB
+var SysSeekCREATE sync.Map
+
+func init() {
+	// 连接数据库
+	var err error
+	Db, err = gorm.Open("mysql", conf.MysqlServer_Username+":"+conf.MysqlServer_Password+"@tcp("+conf.MysqlServer_UrlPort+")/"+conf.MysqlServer_Database+"?charset=utf8mb4&loc=Local&parseTime=True")
+	if err != nil {
+		log.Fatal(err)
+	}
+	// SetMaxIdleConns 设置空闲连接池中的最大连接数。
+	Db.DB().SetMaxIdleConns(10)
+
+	// SetMaxOpenConns 设置数据库连接最大打开数。
+	Db.DB().SetMaxOpenConns(600)
+
+	// SetConnMaxLifetime 设置可重用连接的最长时间
+	Db.DB().SetConnMaxLifetime(time.Hour)
+	//Db.AutoMigrate(&DeviceWarning{})
+	// 开启 Logger, 以展示详细的日志
+	Db.LogMode(true)
+	Db.SetLogger(log.New(logs.LogxOrm, "\r\nWarningBack:", 0))
+
+}
 
 //func Close() {
 //	Db.Close()
@@ -56,38 +67,36 @@ package Warning
 //}
 
 // 创建备份报警表
-//func SeekCREATE(name string) {
-//	_, ok := SysSeekCREATE.Load(name)
-//	if ok {
-//		return
-//	}
-//	rows, _ := Db.Raw("select * from information_schema.TABLES where TABLE_NAME = ? ;", name).Rows() // (*sql.Rows, error)
-//	defer rows.Close()
-//	if !rows.Next() {
-//		logs.Println("没有表：", name, " 现在新建！")
-//		sql := "CREATE TABLE `warning`  (\n                            `ID` bigint NOT NULL AUTO_INCREMENT,\n                            `t_pid` int NULL DEFAULT NULL,\n                            `t_tp` int NULL DEFAULT NULL,\n                            `t_sn` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t__d_name` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t_id` int NULL DEFAULT NULL,\n                            `t__d_s_name` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t__remark` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__ut` datetime NULL DEFAULT NULL,\n                            `t_f_ut` datetime NULL DEFAULT NULL,\n                            `t__text` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__log` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__msid` bigint NULL DEFAULT NULL,\n                            `t__state` int NOT NULL DEFAULT 2,\n                            `create_time` datetime NULL DEFAULT NULL,\n                            `update_time` datetime NULL DEFAULT NULL,\n                            PRIMARY KEY (`ID`) USING BTREE,\n                            INDEX `warning_t__ut`(`t__ut`) USING BTREE,\n                            INDEX `warning_t_pid_t_tp_t__state`(`t_pid`, `t_tp`, `t__state`) USING BTREE,\n                            INDEX `warning_t_sn`(`t_sn`) USING BTREE\n) ENGINE = InnoDB AUTO_INCREMENT = 7522219 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci ROW_FORMAT = Dynamic;"
-//		sql = strings.Replace(sql, "warning", name, -1)
-//		Db.Exec(sql)
-//
-//		time.Sleep(time.Second)
-//	}
-//	SysSeekCREATE.Store(name, "")
-//	return
-//}
+func SeekCREATE(name string) {
+	_, ok := SysSeekCREATE.Load(name)
+	if ok {
+		return
+	}
+	rows, _ := Db.Raw("select * from information_schema.TABLES where TABLE_NAME = ? ;", name).Rows() // (*sql.Rows, error)
+	defer rows.Close()
+	if !rows.Next() {
+		logs.Println("没有表：", name, " 现在新建！")
+		sql := "CREATE TABLE `warning`  (\n                            `ID` bigint NOT NULL AUTO_INCREMENT,\n                            `t_pid` int NULL DEFAULT NULL,\n                            `t_tp` int NULL DEFAULT NULL,\n                            `t_sn` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t__d_name` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t_id` int NULL DEFAULT NULL,\n                            `t__d_s_name` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL,\n                            `t__remark` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__ut` datetime NULL DEFAULT NULL,\n                            `t_f_ut` datetime NULL DEFAULT NULL,\n                            `t__text` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__log` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL,\n                            `t__msid` bigint NULL DEFAULT NULL,\n                            `t__state` int NOT NULL DEFAULT 2,\n                            `create_time` datetime NULL DEFAULT NULL,\n                            `update_time` datetime NULL DEFAULT NULL,\n                            PRIMARY KEY (`ID`) USING BTREE,\n                            INDEX `warning_t__ut`(`t__ut`) USING BTREE,\n                            INDEX `warning_t_pid_t_tp_t__state`(`t_pid`, `t_tp`, `t__state`) USING BTREE,\n                            INDEX `warning_t_sn`(`t_sn`) USING BTREE\n) ENGINE = InnoDB AUTO_INCREMENT = 7522219 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci ROW_FORMAT = Dynamic;"
+		sql = strings.Replace(sql, "warning", name, -1)
+		Db.Exec(sql)
+
+		time.Sleep(time.Second)
+	}
+	SysSeekCREATE.Store(name, "")
+	return
+}
 
 // 插入 备份报警表
-//func INSERT(name string, ID int64) {
-//	SeekCREATE(name) // 创建备份报警表
-//	logs.Println("导入数据：", name, " ID:", ID)
-//	Db.Exec("INSERT INTO "+name+" SELECT * FROM warning WHERE ID = ? ;", ID)
-//
-//}
-//
-//func UPDATE(name string, ID int64, t__log string) {
-//	logs.Println("导入数据：", name, " ID:", ID)
-//	Db.Exec("UPDATE "+name+" SET t__log = ?  WHERE ID = ? ;", t__log, ID)
-//
-//}
+func INSERT(name string, ID int64) {
+	SeekCREATE(name) // 创建备份报警表
+	logs.Println("导入数据：", name, " ID:", ID)
+	Db.Exec("INSERT INTO "+name+" SELECT * FROM warning WHERE ID = ? ;", ID)
+}
+
+func UPDATE(name string, ID int64, t__log string) {
+	logs.Println("导入数据：", name, " ID:", ID)
+	Db.Exec("UPDATE "+name+" SET t__log = ?  WHERE ID = ? ;", t__log, ID)
+}
 
 //func DELETE(name string, currentTime string) {
 //	fmt.Println("删除数据：", name, currentTime)

models/Warning/WarningDB.go

@@ -1,7 +1,6 @@
 package Warning
 
 import (
-	"Cold_mqtt/logs"
 	"database/sql"
 	"fmt"
 	"gorm.io/driver/mysql"
@@ -135,10 +134,10 @@ CreateSQL:
 	}
 }
 
-func UPDATE(name string, ID int64, t__log string) {
-	logs.Println("修改数据：", name, " ID:", ID)
-	db.Exec("UPDATE "+name+" SET t__log = ?  WHERE ID = ? ;", t__log, ID)
-}
+//func UPDATE(name string, ID int64, t__log string) {
+//	logs.Println("修改数据：", name, " ID:", ID)
+//	db.Exec("UPDATE "+name+" SET t__log = ?  WHERE ID = ? ;", t__log, ID)
+//}
 
 func generateMonthArray(startDateStr string, endDateStr string) []string {
 	startDate, _ := time.Parse("2006-01-02 15:04:05", startDateStr)