cicv_data_closed_loop/aarch64/jili/control/main.go

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
}