重新整理物联网服务，分离业务逻辑。

iot/EdgeAgent/main/business_service.go

@@ -0,0 +1,209 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"os/exec"
+	"sentinel/pkg/config"
+	"sentinel/pkg/log"
+	model2 "sentinel/pkg/model"
+	"sentinel/pkg/utils"
+	"strconv"
+	"time"
+)
+
+type BusinessService struct {
+	mqtt          *MQTTService
+	deviceID      string
+	deptId        string
+	cmdTopic      string
+	deviceType    string
+	subscriptions map[string]struct{} // 记录已订阅 topic
+}
+
+func NewBusinessService(m *MQTTService, deviceID string) *BusinessService {
+	return &BusinessService{
+		mqtt:          m,
+		deviceID:      deviceID,
+		subscriptions: make(map[string]struct{}),
+	}
+}
+
+// SubscribeTopic 订阅指定 topic，并记录可取消
+func (b *BusinessService) SubscribeTopic(topic string, qos byte) error {
+	if err := b.mqtt.Subscribe(topic, qos); err != nil {
+		return err
+	}
+	b.subscriptions[topic] = struct{}{}
+	return nil
+}
+
+func getInitTopic(deviceID string) string {
+	return "+/+/+/" + deviceID + "/#"
+}
+func (b *BusinessService) getOwnTopic(deviceID string) string {
+	return b.deptId + "/cmd/" + b.deviceType + "/" + deviceID + "/#"
+}
+
+func (b *BusinessService) Start() error {
+	if err := b.SubscribeTopic(getInitTopic(b.deviceID), 1); err != nil {
+		return err
+	}
+
+	b.mqtt.SetMessageHandler(b.onMQTTMessage)
+
+	// 第一次连接就发送状态信息
+	b.SendStatusInfo()
+
+	return nil
+}
+
+// UnsubscribeTopic 取消订阅指定 topic
+func (b *BusinessService) UnsubscribeTopic(topic string) error {
+	token := b.mqtt.client.Unsubscribe(topic)
+	if token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	delete(b.subscriptions, topic)
+	return nil
+}
+
+// UnsubscribeAll 取消所有已订阅 topic
+func (b *BusinessService) UnsubscribeAll() {
+	for topic := range b.subscriptions {
+		_ = b.mqtt.client.Unsubscribe(topic)
+		delete(b.subscriptions, topic)
+	}
+}
+
+// 消息处理
+func (b *BusinessService) onMQTTMessage(topic string, payload []byte) {
+	model := model2.FromStringToMqttTopic(topic)
+	// 指令
+	if model.Domain == "cmd" && model.DeviceType == b.deviceType {
+		log.Println("收到指令：", model.Resource)
+		switch model.Resource {
+		case "ping":
+			log.Println("pong")
+		case "shutdown":
+			b.handleShutdown()
+		case "restart":
+			b.handleRestart()
+		case "check_update":
+			b.handleCheckUpdate()
+		default:
+			log.Println("未知的命令：", model.Resource)
+		}
+	} else if model.Domain == "status" && model.Resource == "receipt" {
+		b.deviceType = model.DeviceType
+		b.deptId = model.DeptId
+		// 取消订阅之前的初始化主题
+		if b.UnsubscribeTopic(getInitTopic(b.deviceID)) != nil {
+			log.Error("无法取消初始化主题")
+			return
+		}
+		// 新订阅属于自己的主题
+		if b.SubscribeTopic(b.getOwnTopic(b.deviceID), 1) != nil {
+			log.Error("无法定于属于自己的主题")
+			return
+		}
+		log.Println("设备初始化成功：所属项目：", model.DeptId, "\t设备类型：", model.DeviceType)
+	}
+}
+
+func (b *BusinessService) SendStatusInfo() {
+	info := map[string]interface{}{
+		"version":      config.APP_VERSION,
+		"online":       true,
+		"ip":           utils.GetLocalIP(),
+		"hostname":     utils.GetHostname(),
+		"mac":          utils.GetMacAddress(),
+		"os":           utils.GetOSInfo(),
+		"cpu":          utils.GetCPUInfo(),
+		"memory_total": utils.GetMemory(),
+		"disk_total":   utils.GetDisk(),
+		"last_seen":    time.Now().UTC().Format(time.RFC3339),
+	}
+
+	payload, _ := json.Marshal(info)
+	topic := "x/status/x/" + b.deviceID + "/info"
+	qos := byte(1)
+	retained := true
+	log.Println("发送消息：", topic)
+
+	if err := b.mqtt.Publish(topic, qos, retained, payload); err != nil {
+		log.Println("发送状态信息出错：", err)
+	} else {
+		log.Println("发送状态信息：", string(payload))
+	}
+}
+
+// 关闭程序（立即退出）
+func (b *BusinessService) handleShutdown() {
+	os.Exit(0)
+}
+
+// 重启程序
+func (b *BusinessService) handleRestart() {
+	exe, _ := os.Executable()
+	cmd := exec.Command(exe)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	_ = cmd.Start()
+	os.Exit(0)
+}
+func (b *BusinessService) handleCheckUpdate() {
+
+	args := []string{
+		"--version", strconv.Itoa(config.APP_VERSION),
+	}
+
+	launcher := newUpdaterLauncher()
+
+	if err := launcher.Start(args); err != nil {
+		log.Println("[BUS] failed to start updater:", err)
+		return
+	}
+
+	log.Println(
+		"[BUS] updater started, exiting main program",
+	)
+
+	time.Sleep(500 * time.Millisecond)
+	os.Exit(0)
+}
+
+// handleCheckUpdate 触发更新流程（主程序侧）
+//func (b *BusinessService) handleCheckUpdate() {
+//
+//	args := []string{
+//		"--version", strconv.Itoa(config.APP_VERSION),
+//	}
+//
+//	cmd := exec.Command("./updater.exe", args...)
+//	cmd.Stdout = os.Stdout
+//	cmd.Stderr = os.Stderr
+//
+//	// OS 级脱离父进程
+//	switch runtime.GOOS {
+//	case "windows":
+//		cmd.SysProcAttr = &syscall.SysProcAttr{
+//			CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP,
+//		}
+//	}
+//
+//	if err := cmd.Start(); err != nil {
+//		log.Println("[BUS] failed to start updater:", err)
+//		return
+//	}
+//
+//	log.Println(
+//		"[BUS] updater started (pid=%d), exiting main program\n",
+//		cmd.Process.Pid,
+//	)
+//
+//	// 给 updater 留出启动窗口（尤其是 systemd / docker 环境）
+//	time.Sleep(500 * time.Millisecond)
+//
+//	os.Exit(0)
+//}

iot/EdgeAgent/main/main.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"sentinel/pkg/config"
+	"sentinel/pkg/device"
+	"sentinel/pkg/log"
+)
+
+func main() {
+	banner := `
+==========================================================================
+
+██████ █████   █████ ██████       ██    █████ ██████ ██   ██ ██████ 
+██     ██  ██ ██     ██          ████  ██     ██     ███  ██   ██   
+█████  ██   ██ ██  ██ █████      ██  ██ ██  ██ █████  ████ ██   ██   
+██     ██   ██ ██   ██ ██         ██████ ██   ██ ██     ██ ████   ██   
+██████ █████    █████ ██████     ██  ██  █████ ██████ ██   ██   ██   
+                                                                   
+==========================================================================
+`
+	fmt.Println(banner)
+	deviceID := device.GetDeviceID()
+	log.Init(config.Log_file_dic)            // 初始化日志目录
+	log.Info("Device id: " + deviceID)       // 第一次启动记录
+	log.Println("版本号: ", config.APP_VERSION) // 第一次启动记录
+
+	var mqttSvc *MQTTService
+	firstFail := true // 标记是否第一次失败
+	for {
+		mqttSvc = NewMQTTService(config.MQTT_BROKER, deviceID, deviceID, config.PASSWORD, 60)
+		err := mqttSvc.Connect()
+		if err != nil {
+			if firstFail {
+				log.Error("物联网服务连接失败，如未注册设备，请先注册: " + deviceID)
+				firstFail = false
+			}
+			time.Sleep(3 * time.Second) // 5秒后重试
+			continue
+		}
+		break
+	}
+	defer mqttSvc.Close()
+
+	biz := NewBusinessService(mqttSvc, deviceID)
+	for {
+		// MQTT业务
+		err := biz.Start()
+		if err != nil {
+			log.Error("business service start failed: " + err.Error())
+			fmt.Println("业务启动失败，5秒后重试...")
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		break
+	}
+
+	// 主线程循环，可做心跳或状态上报
+	for {
+		time.Sleep(10 * time.Second)
+	}
+}

iot/EdgeAgent/main/main_factory.go

@@ -0,0 +1,7 @@
+package main
+
+import "sentinel/pkg/platform"
+
+func newUpdaterLauncher() platform.UpdaterLauncher {
+	return &updaterLauncher{}
+}

iot/EdgeAgent/main/main_linux.go

@@ -0,0 +1,20 @@
+//go:build linux
+// +build linux
+
+package main
+
+import (
+	"os"
+	"os/exec"
+)
+
+type updaterLauncher struct{}
+
+func (u *updaterLauncher) Start(args []string) error {
+	cmd := exec.Command("./updater", args...)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	// Linux 下先不做进程组处理，保证能跑
+	return cmd.Start()
+}

iot/EdgeAgent/main/main_windows.go

@@ -0,0 +1,24 @@
+//go:build windows
+// +build windows
+
+package main
+
+import (
+	"os"
+	"os/exec"
+	"syscall"
+)
+
+type updaterLauncher struct{}
+
+func (u *updaterLauncher) Start(args []string) error {
+	cmd := exec.Command("./updater.exe", args...)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP,
+	}
+
+	return cmd.Start()
+}

iot/EdgeAgent/main/mqtt_service.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"crypto/tls"
+	"sentinel/pkg/log"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type MQTTService struct {
+	client  mqtt.Client
+	handler func(topic string, payload []byte)
+}
+
+func NewMQTTService(broker, clientID, username, password string, keepalive int) *MQTTService {
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(broker)
+	opts.SetClientID(clientID)
+	opts.SetUsername(username)
+	opts.SetPassword(password)
+	opts.SetKeepAlive(time.Duration(keepalive) * time.Second)
+	opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
+
+	ms := &MQTTService{}
+	opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) {
+		if ms.handler != nil {
+			ms.handler(m.Topic(), m.Payload())
+		}
+	})
+	opts.OnConnect = func(c mqtt.Client) {
+		log.Println("物联网服务已连接")
+	}
+	opts.OnConnectionLost = func(c mqtt.Client, err error) {
+		log.Println("物联网服务已断开：", err)
+	}
+
+	ms.client = mqtt.NewClient(opts)
+	return ms
+}
+
+func (m *MQTTService) Connect() error {
+	token := m.client.Connect()
+	if token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	return nil
+}
+
+func (m *MQTTService) Subscribe(topic string, qos byte) error {
+	token := m.client.Subscribe(topic, qos, nil)
+	if token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	log.Println("物联网服务消息订阅：", topic)
+	return nil
+}
+
+func (m *MQTTService) Publish(topic string, qos byte, retained bool, payload interface{}) error {
+	token := m.client.Publish(topic, qos, retained, payload)
+	token.Wait()
+	return token.Error()
+}
+
+func (m *MQTTService) SetMessageHandler(h func(topic string, payload []byte)) {
+	m.handler = h
+}
+
+func (m *MQTTService) Close() {
+	if m.client != nil {
+		m.client.Disconnect(250)
+	}
+}