ipole_edge_server/edge/ipole/main.go

package main

import (
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/exec"
	"os/signal"
	"runtime"
	"syscall"
	"time"

	jsoniter "github.com/json-iterator/go"
	"github.com/sanbornm/go-selfupdate/selfupdate"
	"github.com/thinkgos/timing/v3"

	"lc/common/util"
)

var json = jsoniter.ConfigCompatibleWithStandardLibrary

var gopool *util.Pool
var IDGen util.IdWorker

var version = "1.4.1"
var appname = "ipole"

func Stat(args ...interface{}) interface{} {
	lastGC := time.Now()
	for {
		if time.Now().Sub(lastGC) > time.Minute {
			timing.Count()
			runtime.GC()
			lastGC = time.Now()
		}
		time.Sleep(time.Second)
	}
	return 0
}

func GetNextUint64() uint64 {
	u64, err := IDGen.NextId()
	if err != nil {
		util.GetTagLog().Errorf("sys", "IDGen.NextId发生错误:%s", err.Error())
		u64 = util.MlNow().Unix()
	}
	return uint64(u64)
}

func EnableUpdate() {
	var updater = &selfupdate.Updater{
		CurrentVersion: version,
		ApiURL:         appConfig.UpdateUrl,
		BinURL:         appConfig.UpdateUrl,
		DiffURL:        appConfig.UpdateUrl,
		Dir:            "update/",
		CmdName:        appname, // app name
	}
	if updater != nil {
		go updater.BackgroundRun()
	}
}

// WatchGoforever 每十分钟查看一次goforever是否启动 hxz
func WatchGoforever(args ...interface{}) interface{} {
	for {
		time.Sleep(1 * time.Minute)

		isRun, err := CheckProRunning("goforever")
		if err != nil {
			util.GetTagLog().Errorf("sys", "检查goforever命令失败:%s", err.Error())
		} else {
			if !isRun {
				//这里重启goforever
				//err := exec.Command("/usr/app/goforever/goforever", "&").Run()
				//err := exec.Command("/bin/sh", "-c", "/usr/app/goforever/goforever ./goforever &").Run()
				err := exec.Command("/bin/sh", "-c", "/usr/app/goforever ./goforever &").Run()
				if err != nil {
					util.GetTagLog().Errorf("sys", "重启goforever出错: %s", err.Error())
				} else {
					util.GetTagLog().Info("sys", "重启goforever成功,当前时间:", util.MlNow().String())
				}
			} else {
				util.GetTagLog().Info("sys", "goforever进程正在运行:", util.MlNow().String())
			}
		}
	}
}

// SyncTime 每半小时同步一次时间
func SyncTime(args ...interface{}) interface{} {
	for {
		if appConfig.NtpServer == "" {
			time.Sleep(30 * time.Minute)
			continue
		}
		err := exec.Command("ntpclient", "-h", appConfig.NtpServer, "-s").Run()
		if err != nil {
			util.GetTagLog().Errorf("sys", "时间同步失败:%s", err.Error())
		} else {
			util.GetTagLog().Info("sys", "时间同步成功,当前时间:", util.MlNow().String())
		}
		time.Sleep(30 * time.Minute)
	}
}

func SignalProcess(args ...interface{}) interface{} {
	for {
		ch := make(chan os.Signal)
		signal.Notify(ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGKILL, syscall.SIGQUIT)

		util.GetTagLog().Info("sys", "检测到退出信号:", <-ch)

	}
	return 0
}

// func HttpProcess(w http.ResponseWriter, r *http.Request) {

// 	if GetMonitorStatus().GetMonitorRunStatus() {
// 		fmt.Fprintf(w, "1")
// 	} else {
// 		fmt.Fprintf(w, "0")
// 	}
// }

func main() {
	runtime.GOMAXPROCS(runtime.NumCPU() * 2)

	util.InitLogrus("release")

	go func() {
		util.GetTagLog().Info("sys", http.ListenAndServe(":9999", nil))
	}()

	util.GetTagLog().Infof("sys", "当前程序版本:%s", appname+" "+version)

	if err := loadAppConfig(); err != nil {
		util.GetTagLog().Errorf("sys", "loadAppConfig错误:%s", err.Error())
		return
	}
	//升级检查
	if appConfig.Upgrade {
		EnableUpdate()
	}

	if err := loadSerialConfig(); err != nil {
		util.GetTagLog().Errorf("sys", "loadAppConfig错误:%s", err.Error())
		return
	}

	IDGen.InitIdWorker(1000, 1)

	GetMQTTMgr()

	InitRedis(&util.RedisConfig{Redis_Address: appConfig.Edge.Redis.Server, Redis_Password: appConfig.Edge.Redis.Password})
	defer UninitRedis()

	InitCloudMqttSubscribeTopics()

	//初始化日志配置（加载持久化级别或使用默认值）
	InitLogCfg()

	// 启动健康检测（检测部署标记文件）
	go StartHealthCheckIfNeeded()

	//打开串口
	GetSerialMgr().AddSerialPorts(serialConfig.Serial)

	//创建设备并启动数据处理流程
	for k := range serialConfig.Serial {
		devinfos, err := LoadDev(k)
		if err != nil {
			util.GetTagLog().Warnf("sys", "加载串口[%d]的设备配置文件dev/%d.json失败: %v ,该串口无设备将被跳过", k, k, err)
			continue
		}
		GetDeviceMgr().AddDevices(devinfos)
	}

	// 启动本地部署 HTTP 服务（端口9998）
	StartDeployHTTP()

	gopool = util.NewPool(10)
	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
		gopool.Add(fn, args)
	})
	gopool.Add(SyncTime, 0)
	// gopool.Add(WatchGoforever, 0)
	// gopool.Add(SignalProcess, 8)
	gopool.Add(WatchDevConfig, 1)
	gopool.Add(WatchModelConfig, 2)
	gopool.Add(WatchConfConfig, 3)
	gopool.Add(GetMQTTMgr().MQTTMessageHandle, 4)
	gopool.Add(GetMQTTMgr().MQTTConnectMgr, 5)
	gopool.Add(Stat, 6)
	gopool.Add(RadarReceive, 7)
	gopool.Add(CleanLogs, 9)
	gopool.Add(Heartbeat, 10)
	gopool.Run()
	gopool.Wait()
}