|
|
@@ -5,9 +5,11 @@ import (
|
|
|
commonService "cicv-data-closedloop/aarch64/jili/common/service"
|
|
|
"cicv-data-closedloop/common/config/c_log"
|
|
|
"cicv-data-closedloop/common/util"
|
|
|
+ "fmt"
|
|
|
"net/rpc"
|
|
|
"os"
|
|
|
"runtime"
|
|
|
+ "strings"
|
|
|
"time"
|
|
|
)
|
|
|
|
|
|
@@ -32,80 +34,78 @@ func main() {
|
|
|
}
|
|
|
|
|
|
func RunWithInternet() {
|
|
|
- lastStatus := "NONE"
|
|
|
wait := false
|
|
|
- // 轮询任务接口判断是否有更新
|
|
|
+ // 轮询任务接口, 决定数采程序启停
|
|
|
for {
|
|
|
if wait { // 第一次就不等待一分钟了
|
|
|
time.Sleep(time.Duration(60) * time.Second)
|
|
|
}
|
|
|
wait = true
|
|
|
- // 1 获取当前设备的任务的 status
|
|
|
- status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
|
|
|
+
|
|
|
+ // 1. 获取当前设备的任务配置
|
|
|
+ platformTasks, err := commonConfig.GetConfig()
|
|
|
+ fmt.Println("platformTasks", platformTasks)
|
|
|
+ commonConfig.PlatformTasks = platformTasks
|
|
|
if err != nil {
|
|
|
- c_log.GlobalLogger.Error("获取配置status失败：", err)
|
|
|
+ c_log.GlobalLogger.Error("获取平台任务失败：", err)
|
|
|
continue
|
|
|
}
|
|
|
- c_log.GlobalLogger.Info("【lastStatus】=", lastStatus, "，【status】=", status)
|
|
|
- // 2 判断 status
|
|
|
- // UN_CHANGE 没有新的任务，无需更改
|
|
|
- // CHANGE 有新的任务，需要杀死旧的任务并重启
|
|
|
- // NONE 设备没有配置任务，需要杀死旧的任务
|
|
|
- if status == "UN_CHANGE" {
|
|
|
- lastStatus = "UN_CHANGE"
|
|
|
- continue
|
|
|
- } else if status == "CHANGE" || status == "NONE" {
|
|
|
- if lastStatus == "CHANGE" && status == "CHANGE" { // 供更新使用
|
|
|
- commonConfig.InitPlatformConfig()
|
|
|
- continue
|
|
|
- }
|
|
|
- if lastStatus == "NONE" && status == "NONE" {
|
|
|
+
|
|
|
+ // 2. 判断当前时间是否有需要执行的任务
|
|
|
+ for _, task := range platformTasks {
|
|
|
+ flag := shouldExecuteTask(task.StartTime, task.EndTime)
|
|
|
+ fmt.Println("flag", flag)
|
|
|
+ // 2.1 当前时间不执行此任务，跳过
|
|
|
+ if !flag {
|
|
|
continue
|
|
|
}
|
|
|
- // 3 发送rpc信号杀死两个服务，并重启程序
|
|
|
- if lastStatus == "NONE" && status == "CHANGE" {
|
|
|
- startMasterOrSlave()
|
|
|
- lastStatus = status
|
|
|
- c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
|
|
|
- commonConfig.InitPlatformConfig()
|
|
|
- continue
|
|
|
+ // 2.2 当前时间需要执行此任务
|
|
|
+ // 2.2.1 判断此任务是否正在执行
|
|
|
+ if commonConfig.CurrTask.TaskConfigId == task.TaskConfigId { // 此任务正在执行，退出循环
|
|
|
+ break
|
|
|
}
|
|
|
- var killArgs commonService.KillSignal
|
|
|
- //if lastStatus == "UN_CHANGE" && status == "CHANGE" {
|
|
|
- // killArgs = commonService.KillSignal{DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
|
|
|
- // c_log.GlobalLogger.Info("更新任务，发送rpc重启信号到本地：", killArgs)
|
|
|
- //}
|
|
|
- //if lastStatus == "UN_CHANGE" && status == "NONE" {
|
|
|
- // killArgs = commonService.KillSignal{DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
|
|
|
- // c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地：", killArgs)
|
|
|
- //}
|
|
|
-
|
|
|
- KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
|
|
|
- if err != nil {
|
|
|
- // 此处如果连接失败说明采集程序已经停止了
|
|
|
- lastStatus = "NONE"
|
|
|
- c_log.GlobalLogger.Error("采集程序已经停止：", err)
|
|
|
- continue
|
|
|
+ // 此任务没有在执行，需要启动此任务
|
|
|
+ // 2.2.2 判断是否有其他任务正在执行
|
|
|
+ if commonConfig.CurrTask.TaskConfigId != "" { // 存在其他任务正在执行，杀死旧的任务
|
|
|
+ c_log.GlobalLogger.Info("需要执行的任务id为", task.TaskConfigId, ", 当前任务id为：", commonConfig.CurrTask.TaskConfigId, ", 需要杀死该任务。")
|
|
|
+ // 杀死旧的任务
|
|
|
+ var killArgs commonService.KillSignal
|
|
|
+ // todo 吉利新平台没有DropUploadData字段（旧平台，已采数据：保留/丢弃）
|
|
|
+ killArgs = commonService.KillSignal{DropUploadData: true, Restart: false}
|
|
|
+ c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地：", killArgs)
|
|
|
+ KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
|
|
|
+ if err != nil {
|
|
|
+ // 此处如果连接失败说明采集程序已经停止了
|
|
|
+ commonConfig.RecordTopics = []string{}
|
|
|
+ commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
+ c_log.GlobalLogger.Error("采集程序已经停止：", err)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ reply := 0
|
|
|
+ if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
|
|
|
+ c_log.GlobalLogger.Error("发送rpc请求到master失败：", err)
|
|
|
+ // 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
|
|
|
+ //KillRpcClient.Close()
|
|
|
+ //continue
|
|
|
+ }
|
|
|
+ c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
|
|
|
+ commonConfig.RecordTopics = []string{}
|
|
|
+ commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
+ if err = KillRpcClient.Close(); err != nil {
|
|
|
+ // 不做处理
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- reply := 0
|
|
|
- if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
|
|
|
- c_log.GlobalLogger.Error("发送rpc请求到master失败：", err)
|
|
|
- // 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
|
|
|
- //KillRpcClient.Close()
|
|
|
- //continue
|
|
|
- }
|
|
|
- lastStatus = status
|
|
|
- c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
|
|
|
- commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
|
|
|
- if err = KillRpcClient.Close(); err != nil {
|
|
|
- // 不做处理
|
|
|
- }
|
|
|
- } else {
|
|
|
- c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
|
|
|
+ // 开启新的任务
|
|
|
+ startMasterOrSlave()
|
|
|
+ commonConfig.RecordTopics = getUniqueCollectionTopics(task.TaskTriggers)
|
|
|
+ commonConfig.CurrTask = task
|
|
|
+ // 退出循环
|
|
|
+ break
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
func RunWithoutInternet() {
|
|
|
startMasterOrSlave()
|
|
|
}
|
|
|
@@ -117,3 +117,57 @@ func startMasterOrSlave() {
|
|
|
}
|
|
|
c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
|
|
|
}
|
|
|
+
|
|
|
+// shouldExecuteTask 给定任务起止时间，判断当前时刻此任务是否应该执行
|
|
|
+func shouldExecuteTask(startTimeStr, endTimeStr string) bool {
|
|
|
+ // 定义时间格式（根据输入的时间字符串格式）
|
|
|
+ timeFormat := "2006-01-02 15:04:05"
|
|
|
+
|
|
|
+ // 注意： time.Parse 时间没有考虑时区，当做默认的 +0 时区 UTC 时间来解析了，就造成了实际时间比预期的 +8 小时
|
|
|
+ // 解析 startTime 和 endTime
|
|
|
+ startTime, err := time.ParseInLocation(timeFormat, startTimeStr, time.Local)
|
|
|
+ //fmt.Println("startTime", startTime)
|
|
|
+ if err != nil {
|
|
|
+ fmt.Println("Error parsing startTime:", err)
|
|
|
+ return false
|
|
|
+ }
|
|
|
+ endTime, err := time.ParseInLocation(timeFormat, endTimeStr, time.Local)
|
|
|
+ //fmt.Println("endTime", endTime)
|
|
|
+ if err != nil {
|
|
|
+ fmt.Println("Error parsing endTime:", err)
|
|
|
+ return false
|
|
|
+ }
|
|
|
+
|
|
|
+ // 获取当前时间
|
|
|
+ currentTime := time.Now()
|
|
|
+ //fmt.Println("currentTime", currentTime)
|
|
|
+
|
|
|
+ // 判断当前时间是否在 startTime 和 endTime 之间
|
|
|
+ return !currentTime.Before(startTime) && !currentTime.After(endTime)
|
|
|
+}
|
|
|
+
|
|
|
+// getUniqueCollectionTopics 统计需要采集的 topic 并去重
|
|
|
+func getUniqueCollectionTopics(taskTriggers []commonConfig.TaskTrigger) []string {
|
|
|
+ // 使用 map 去重
|
|
|
+ topicMap := make(map[string]struct{})
|
|
|
+
|
|
|
+ // 遍历 taskTriggers
|
|
|
+ for _, trigger := range taskTriggers {
|
|
|
+ // 分割 collectionTopic 字段
|
|
|
+ topics := strings.Split(trigger.CollectionTopic, ",")
|
|
|
+ for _, topic := range topics {
|
|
|
+ trimmedTopic := strings.TrimSpace(topic) // 去除空格
|
|
|
+ if trimmedTopic != "" { // 忽略空字符串
|
|
|
+ topicMap[trimmedTopic] = struct{}{}
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 将 map 中的 key 转换为切片
|
|
|
+ var uniqueTopics []string
|
|
|
+ for topic := range topicMap {
|
|
|
+ uniqueTopics = append(uniqueTopics, topic)
|
|
|
+ }
|
|
|
+
|
|
|
+ return uniqueTopics
|
|
|
+}
|
