cicv_data_closed_loop/kinglong/common/svc/kill_self.go

package svc

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

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)
	ChannelKillRosRecord <- 1
	ChannelKillWindowProducer <- 1
	if args.DropUploadData == true {
		// 3-1 等待上传结束再杀死
		ChannelKillMove <- 1
		ChannelKillConsume <- 1
	} else {
		// 3-2 直接杀死
		ChannelKillMove <- 2
		ChannelKillConsume <- 2
	}
	go killDone()
	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":
		log.GlobalLogger.Infof("已杀死record打包goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
		ChannelKillDiskClean <- 1
		KillTimes++
		close(ChannelKillRosRecord)
	case "2":
		log.GlobalLogger.Infof("已杀死bag包数量维护goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
		KillTimes++
		close(ChannelKillDiskClean)
	case "3":
		log.GlobalLogger.Infof("已杀死时间窗口生产者，当前自杀进度 %v / %v", KillTimes, KillChannel)
		KillTimes++
		close(ChannelKillWindowProducer)
	case "4":
		log.GlobalLogger.Infof("已杀死bag包移动goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
		KillTimes++
		close(ChannelKillMove)
	case "5":
		log.GlobalLogger.Infof("已杀死bag包消费goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
		KillTimes++
		close(ChannelKillConsume)
	}
	MutexKill.Unlock()
}

func killDone() {
	log.GlobalLogger.Infof("自杀完毕，启动新的程序。")
	if KillChannel == KillTimes {
		if _, err := util.ExecuteWithPath(cfg.LocalConfig.RestartDir, cfg.LocalConfig.RestartCmd); err != nil {
			log.GlobalLogger.Error("启动新的程序失败：", err)
		}
		os.Exit(0)
	}
}