cicv_data_closed_loop/pjisuv/common/service/kill_self.go

package svc

import (
	"cicv-data-closedloop/kinglong/common/cfg"
	"cicv-data-closedloop/kinglong/common/log"
	"cicv-data-closedloop/kinglong/common/util"
	commonConfig "cicv-data-closedloop/pji/common/cfg"
	"net/rpc"
	"os"
	"sync"
	"time"
)

var (
	ChannelKillRosRecord      = make(chan int)
	ChannelKillDiskClean      = make(chan int)
	ChannelKillWindowProducer = make(chan int)
	ChannelKillMove           = make(chan int)
	ChannelKillConsume        = make(chan int)

	KillChannel = 5
	KillTimes   = 0
	MutexKill   sync.Mutex
)

// KillSignal 停止信号，主从节点接收到数据后准备重启
type KillSignal struct {
	NodeName       string
	DropUploadData bool
	Restart        bool
}

// KillService 定义要远程调用的类型和方法
type KillService struct{}

// Kill 杀死自身程序，通过通道实现 方法必须满足RPC规范：函数有两个参数，第一个参数是请求，第二个是响应
func (m *KillService) Kill(args *KillSignal, reply *int) error {
	log.GlobalLogger.Info("接收到自杀信号：", *args)
	// 1 杀死 rosbag record 命令
	ChannelKillRosRecord <- 1
	// 2 杀死所有 ros 订阅者
	ChannelKillWindowProducer <- 1
	if args.DropUploadData == true {
		// 3-1 等待上传结束再杀死
		ChannelKillMove <- 1
		ChannelKillConsume <- 1
	} else {
		// 3-2 直接杀死
		ChannelKillMove <- 2
		ChannelKillConsume <- 2
	}
	go killDone(args.Restart)
	return nil
}
func WaitKillSelf() {
	killService := new(KillService)
	if err := rpc.Register(killService); err != nil {
		log.GlobalLogger.Error("注册rpc服务失败：", err)
		return
	}

	// 等待并处理远程调用请求
	for {
		conn, err := cfg.KillSignalListener.Accept()
		if err != nil {
			continue
		}
		go rpc.ServeConn(conn)
	}
}

func AddKillTimes(info string) {
	MutexKill.Lock()

	switch info {
	case "1":
		close(ChannelKillRosRecord)
		KillTimes++
		log.GlobalLogger.Infof("已杀死record打包goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
		ChannelKillDiskClean <- 1
	case "2":
		close(ChannelKillDiskClean)
		KillTimes++
		log.GlobalLogger.Infof("已杀死bag包数量维护goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
	case "3":
		close(ChannelKillWindowProducer)
		KillTimes++
		log.GlobalLogger.Infof("已杀死时间窗口生产者，当前自杀进度 %v / %v", KillTimes, KillChannel)
	case "4":
		close(ChannelKillMove)
		KillTimes++
		log.GlobalLogger.Infof("已杀死bag包移动goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
	case "5":
		close(ChannelKillConsume)
		KillTimes++
		log.GlobalLogger.Infof("已杀死bag包消费goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
	}
	MutexKill.Unlock()
}

func killDone(restart bool) {
	for {
		time.Sleep(time.Duration(1) * time.Second)
		if KillChannel == KillTimes {
			if restart {
				_, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...)
				if err != nil {
					log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
					os.Exit(-1)
				}
				log.GlobalLogger.Info("数据采集任务更新，正常退出当前程序。")
			} else {
				log.GlobalLogger.Info("数据采集任务终止，正常退出当前程序。")
			}
			os.Exit(0)
		}
	}
}