package main import ( commonConfig "cicv-data-closedloop/aarch64/pjibot_guide/common/config" commonService "cicv-data-closedloop/aarch64/pjibot_guide/common/service" "cicv-data-closedloop/aarch64/pjibot_guide/common/variable" "cicv-data-closedloop/aarch64/pjibot_guide/control/pkg" "cicv-data-closedloop/common/config/c_log" "cicv-data-closedloop/common/entity" "cicv-data-closedloop/common/util" "fmt" "net/rpc" "os" "runtime" "time" ) var ( applicationName = "pji-control" localStatus = "idle" cloudStatus = "NONE" lastLocalStatus = "idle" lastCloudStatus = "NONE" launchedFlag = false renewedFlag = false configuredFlag = false renewTimer *time.Timer // 续约定时器 RenewDur = 5 // 续约时间为5分钟 ) func init() { runtime.GOMAXPROCS(1) // 初始化日志配置 c_log.InitLog(variable.LogDir, applicationName) // 初始化本地配置文件(第1处配置,在本地文件) commonConfig.InitLocalConfig(variable.LocalConfigPath) // 初始化Oss连接信息 commonConfig.InitOssConfig() // 初始化业务逻辑配置信息,配置文件在oss上(第2处配置,在oss文件) commonConfig.InitCloudConfig() // 初始化rpc客户端,用于杀死旧的采集程序 // 初始化websocket配置 commonConfig.InitWebsocketConfig() } func renew() { if renewTimer != nil { renewTimer.Stop() } renewedFlag = true renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() { renewedFlag = false }) c_log.GlobalLogger.Infof("续约时间【%v】分钟 - 成功。", RenewDur) } func startMasterNode() { c_log.GlobalLogger.Info("获取数据闭环平台最新配置。") commonConfig.InitPlatformConfig() 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) launchedFlag = true c_log.GlobalLogger.Info("数采程序启动 - 成功。") } func stopMasterNode() { // 发送rpc信号杀死采集程序 var killArgs commonService.KillSignal killArgs = commonService.KillSignal{NodeName: "master", 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 { // 此处如果连接失败说明采集程序已经停止了 lastCloudStatus = "NONE" c_log.GlobalLogger.Error("采集程序已经停止:", err) return } 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.PlatformConfig = commonConfig.PlatformConfigStruct{} if err = KillRpcClient.Close(); err != nil { // 不做处理 } launchedFlag = false c_log.GlobalLogger.Info("数采程序关闭 - 成功。") } func main() { localTurnLength := 2 // s cloudTurnLength := 60 // s waitStopLength := 1 // min // 更新本地任务状态 go pkg.GetLocalStatus(&localStatus, &lastLocalStatus, localTurnLength) // 更新云端任务状态 go pkg.GetCloudStatus(&cloudStatus, &lastCloudStatus, cloudTurnLength) for { if launchedFlag { // 当前已启动master节点 time.Sleep(time.Duration(cloudTurnLength) * time.Second) } else { time.Sleep(time.Duration(localTurnLength) * time.Second) } fmt.Println("localStatus: ", localStatus, "lastLocalStatus: ", lastLocalStatus) fmt.Println("cloudStatus: ", cloudStatus, "lastCloudStatus: ", lastCloudStatus) // 综合判断 cloudStatus 和 localStatus // cloudStatus // UN_CHANGE 没有新的任务,无需更改 // CHANGE 有新的任务,需要杀死旧的数采任务并重启 // NONE 设备没有配置任务,需要杀死旧的数采任务 // localStatus // idle 空闲状态,此状态下不启动数采任务 // running 繁忙状态,此状态需要启动数采任务 // error 错误状态,此状态下不启动数采任务 // 本地任务状态负责启停master if localStatus == "running" { if !launchedFlag { // 目前未启动数采程序 c_log.GlobalLogger.Info("数采程序启动 - 开始。") startMasterNode() } else { // 已启动数采程序 c_log.GlobalLogger.Info("设备仍处于运行状态,续约 - 开始。") // 续约 renew() } continue } else if localStatus == "idle" { if !renewedFlag && launchedFlag && len(entity.TimeWindowConsumerQueue) == 0 { c_log.GlobalLogger.Info("设备不在运行状态且没有待处理的数据,数采程序关闭 - 开始。") time.Sleep(time.Duration(waitStopLength) * time.Minute) stopMasterNode() } continue } else if localStatus == "error" { c_log.GlobalLogger.Error("设备运行状态出错,停止数采程序。") stopMasterNode() } else { c_log.GlobalLogger.Error("未知的设备运行状态。【status】=", localStatus) } // 云端任务状态负责更新配置 go pkg.GetCloudConfig(cloudStatus, lastCloudStatus, cloudTurnLength) //if cloudStatus == "UN_CHANGE" { // continue //} else if cloudStatus == "CHANGE" || cloudStatus == "NONE" { // if lastCloudStatus == "CHANGE" && cloudStatus == "CHANGE" { // 供更新使用 // commonConfig.InitPlatformConfig() // continue // } // if lastCloudStatus == "NONE" && cloudStatus == "NONE" { // continue // } // // if lastCloudStatus == "NONE" && cloudStatus == "CHANGE" { // 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) // c_log.GlobalLogger.Info("获取数据闭环平台最新配置。") // commonConfig.InitPlatformConfig() // continue // } // // // 3 发送rpc信号杀死采集程序 // var killArgs commonService.KillSignal // if lastCloudStatus == "UN_CHANGE" && cloudStatus == "CHANGE" { // killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true} // c_log.GlobalLogger.Info("更新任务,发送rpc重启信号:", killArgs) // } // if lastCloudStatus == "UN_CHANGE" && cloudStatus == "NONE" { // killArgs = commonService.KillSignal{NodeName: "master", 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 { // // 此处如果连接失败说明采集程序已经停止了 // lastCloudStatus = "NONE" // 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.PlatformConfig = commonConfig.PlatformConfigStruct{} // if err = KillRpcClient.Close(); err != nil { // // 不做处理 // } //} else { // c_log.GlobalLogger.Error("未知的采集任务状态。【cloudStatus】=", cloudStatus) //} } //select {} }