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/util" "net/rpc" "os" "runtime" "time" ) var applicationName = "pji-control" 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客户端,用于杀死旧的采集程序 } func main() { go judgeState1() go pkg.JudgeLocal() go judgeState3() select {} } func judgeState1() { init := true turnLength := 60 lastStatus := "NONE" // 轮询任务接口判断是否有更新 for { c_log.GlobalLogger.Errorf("一轮次扫描时间【%v】秒:", turnLength) if init { time.Sleep(time.Duration(1) * time.Second) init = false } else { time.Sleep(time.Duration(turnLength) * time.Second) } // 1 获取当前设备的任务的 status status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId) if err != nil { c_log.GlobalLogger.Error("获取配置status失败:", err) continue } // 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" { continue } // 3 发送rpc信号杀死采集程序 if lastStatus == "NONE" && status == "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) lastStatus = status c_log.GlobalLogger.Info("获取数据闭环平台最新配置。") commonConfig.InitPlatformConfig() continue } var killArgs commonService.KillSignal if lastStatus == "UN_CHANGE" && status == "CHANGE" { killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true} c_log.GlobalLogger.Info("更新任务,发送rpc重启信号:", killArgs) } if lastStatus == "UN_CHANGE" && status == "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 { // 此处如果连接失败说明采集程序已经停止了 lastStatus = "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 } lastStatus = status c_log.GlobalLogger.Info("结束任务后,将数据闭环平台配置置空。") commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{} if err = KillRpcClient.Close(); err != nil { // 不做处理 } } else { c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status) } } } func judgeState3() {}