package main import ( commonConfig "cicv-data-closedloop/aarch64/jili/common/config" 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" ) func init() { runtime.GOMAXPROCS(1) // 初始化日志配置 c_log.InitLog(commonConfig.LogDir, commonConfig.ControlLogPrefix) // 初始化本地配置文件(第1处配置,在本地文件) commonConfig.InitLocalConfig() // 初始化Oss连接信息 commonConfig.InitOssConfig() // 初始化业务逻辑配置信息,配置文件在oss上(第2处配置,在oss文件) commonConfig.InitCloudConfig() } func main() { if commonConfig.LocalConfig.Internet { RunWithInternet() } else { RunWithoutInternet() } } func RunWithInternet() { wait := false // 轮询任务接口, 决定数采程序启停 for { if wait { // 第一次就不等待一分钟了 time.Sleep(time.Duration(60) * time.Second) } wait = true // 1. 获取当前设备的任务配置 platformTasks, err := commonConfig.GetConfig() fmt.Println("platformTasks", platformTasks) commonConfig.PlatformTasks = platformTasks if err != nil { c_log.GlobalLogger.Error("获取平台任务失败:", err) continue } // 2. 判断当前时间是否有需要执行的任务 for _, task := range platformTasks { flag := shouldExecuteTask(task.StartTime, task.EndTime) fmt.Println("flag", flag) // 2.1 当前时间不执行此任务,跳过 if !flag { continue } // 2.2 当前时间需要执行此任务 // 2.2.1 判断此任务是否正在执行 if commonConfig.CurrTask.TaskConfigId == task.TaskConfigId { // 此任务正在执行,退出循环 break } // 此任务没有在执行,需要启动此任务 // 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 { // 不做处理 } } // 开启新的任务 startMasterOrSlave() commonConfig.RecordTopics = getUniqueCollectionTopics(task.TaskTriggers) commonConfig.CurrTask = task // 退出循环 break } } } func RunWithoutInternet() { startMasterOrSlave() } func startMasterOrSlave() { if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil { c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, ":", err) os.Exit(-1) } 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 }