main.go 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  1. package main
  2. import (
  3. commonConfig "cicv-data-closedloop/aarch64/pjibot_guide/common/config"
  4. commonService "cicv-data-closedloop/aarch64/pjibot_guide/common/service"
  5. "cicv-data-closedloop/aarch64/pjibot_guide/common/variable"
  6. "cicv-data-closedloop/aarch64/pjibot_guide/control/pkg"
  7. "cicv-data-closedloop/common/config/c_log"
  8. "cicv-data-closedloop/common/util"
  9. "fmt"
  10. "net/rpc"
  11. "os"
  12. "runtime"
  13. "time"
  14. )
  15. var (
  16. applicationName = "pji-control"
  17. localStatus = "idle"
  18. cloudStatus = "NONE"
  19. )
  20. func init() {
  21. runtime.GOMAXPROCS(1)
  22. // 初始化日志配置
  23. c_log.InitLog(variable.LogDir, applicationName)
  24. // 初始化本地配置文件(第1处配置,在本地文件)
  25. commonConfig.InitLocalConfig(variable.LocalConfigPath)
  26. // 初始化Oss连接信息
  27. commonConfig.InitOssConfig()
  28. // 初始化业务逻辑配置信息,配置文件在oss上(第2处配置,在oss文件)
  29. commonConfig.InitCloudConfig()
  30. // 初始化rpc客户端,用于杀死旧的采集程序
  31. // 初始化websocket配置
  32. commonConfig.InitWebsocketConfig()
  33. }
  34. func main() {
  35. localTurnLength := 2
  36. cloudTurnLength := 60
  37. overallTurnLength := localTurnLength
  38. //lastCloudStatus := "NONE"
  39. go pkg.GetLocalStatus(&localStatus, localTurnLength)
  40. go pkg.GetCloudStatus(&cloudStatus, cloudTurnLength)
  41. for {
  42. time.Sleep(time.Duration(overallTurnLength) * time.Second)
  43. fmt.Println("localStatus: ", localStatus)
  44. fmt.Println("cloudStatus: ", cloudStatus)
  45. //if cloudStatus == "UN_CHANGE" {
  46. // lastCloudStatus = "UN_CHANGE"
  47. // continue
  48. //}
  49. }
  50. //select {}
  51. }
  52. func judgeState1() {
  53. init := true
  54. turnLength := 60
  55. lastStatus := "NONE"
  56. // 轮询任务接口判断是否有更新
  57. for {
  58. c_log.GlobalLogger.Errorf("一轮次扫描时间【%v】秒:", turnLength)
  59. if init {
  60. time.Sleep(time.Duration(1) * time.Second)
  61. init = false
  62. } else {
  63. time.Sleep(time.Duration(turnLength) * time.Second)
  64. }
  65. // 1 获取当前设备的任务的 status
  66. status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
  67. if err != nil {
  68. c_log.GlobalLogger.Error("获取配置status失败:", err)
  69. continue
  70. }
  71. // 2 判断 status
  72. // UN_CHANGE 没有新的任务,无需更改
  73. // CHANGE 有新的任务,需要杀死旧的任务并重启
  74. // NONE 设备没有配置任务,需要杀死旧的任务
  75. if status == "UN_CHANGE" {
  76. lastStatus = "UN_CHANGE"
  77. continue
  78. } else if status == "CHANGE" || status == "NONE" {
  79. if lastStatus == "CHANGE" && status == "CHANGE" { // 供更新使用
  80. commonConfig.InitPlatformConfig()
  81. continue
  82. }
  83. if lastStatus == "NONE" && status == "NONE" {
  84. continue
  85. }
  86. if lastStatus == "NONE" && status == "CHANGE" {
  87. if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
  88. c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, ":", err)
  89. os.Exit(-1)
  90. }
  91. c_log.GlobalLogger.Info("启动任务,本地执行启动命令:【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
  92. lastStatus = status
  93. c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
  94. commonConfig.InitPlatformConfig()
  95. continue
  96. }
  97. // 3 发送rpc信号杀死采集程序
  98. var killArgs commonService.KillSignal
  99. if lastStatus == "UN_CHANGE" && status == "CHANGE" {
  100. killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
  101. c_log.GlobalLogger.Info("更新任务,发送rpc重启信号:", killArgs)
  102. }
  103. if lastStatus == "UN_CHANGE" && status == "NONE" {
  104. killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
  105. c_log.GlobalLogger.Info("杀死任务,发送rpc结束信号:", killArgs)
  106. }
  107. KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
  108. if err != nil {
  109. // 此处如果连接失败说明采集程序已经停止了
  110. lastStatus = "NONE"
  111. c_log.GlobalLogger.Error("采集程序已经停止:", err)
  112. continue
  113. }
  114. reply := 0
  115. if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
  116. c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错:", err)
  117. // 这里可能会报错unexpected EOF但是不影响,先注释 close 和 continue
  118. //KillRpcClient.Close()
  119. //continue
  120. }
  121. lastStatus = status
  122. c_log.GlobalLogger.Info("结束任务后,将数据闭环平台配置置空。")
  123. commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
  124. if err = KillRpcClient.Close(); err != nil {
  125. // 不做处理
  126. }
  127. } else {
  128. c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
  129. }
  130. }
  131. }
  132. func judgeState3() {}