移除原心跳配置和业务，改为使用 WebSocket Ping Pong 实现，并增加超时逻辑

cmd/agent/agent.example.yaml

@@ -33,10 +33,6 @@ collector:
   # 建议: 5-60 秒，太短会增加服务器负载
   interval: 5
 
-  # 心跳间隔（秒）
-  # 建议: 30-120 秒
-  heartbeat_interval: 30
-
   # 网络采集包含的网卡列表（白名单，支持正则表达式）
   # 如果配置了此项，则只采集匹配的网卡，忽略 network_exclude
   # 例如: ["^eth0$", "^en0$", "^ens.*"]

cmd/agent/main.go

@@ -485,9 +485,8 @@ func registerAgent(cmd *cobra.Command, args []string) {
 			LogCompress:   true,
 		},
 		Collector: config.CollectorConfig{
-			Interval:          5,
-			HeartbeatInterval: 30,
-			NetworkExclude:    config.DefaultNetworkExcludePatterns(),
+			Interval:       5,
+			NetworkExclude: config.DefaultNetworkExcludePatterns(),
 		},
 		AutoUpdate: config.AutoUpdateConfig{
 			Enabled:       true,
@@ -569,7 +568,6 @@ func showInfo(cmd *cobra.Command, args []string) {
 	// 采集器配置
 	fmt.Println("📊 采集器配置:")
 	fmt.Printf("   采集间隔: %d 秒\n", cfg.Collector.Interval)
-	fmt.Printf("   心跳间隔: %d 秒\n", cfg.Collector.HeartbeatInterval)
 	if len(cfg.Collector.NetworkExclude) > 0 {
 		fmt.Printf("   网卡过滤规则: %v\n", cfg.Collector.NetworkExclude)
 	}

internal/handler/agent_handler.go

@@ -49,6 +49,7 @@ func NewAgentHandler(logger *zap.Logger, agentService *service.AgentService, tra
 
 	// 设置WebSocket消息处理器
 	wsManager.SetMessageHandler(h.handleWebSocketMessage)
+	wsManager.SetPongHandler(h.handleWebSocketPong)
 
 	return h
 }

internal/handler/agent_ws_handler.go

@@ -77,9 +77,6 @@ func (h *AgentHandler) HandleWebSocket(c echo.Context) error {
 // handleWebSocketMessage 处理WebSocket消息
 func (h *AgentHandler) handleWebSocketMessage(ctx context.Context, agentID string, messageType string, data json.RawMessage) error {
 	switch protocol.MessageType(messageType) {
-	case protocol.MessageTypeHeartbeat:
-		return h.handleHeartbeatMessage(ctx, agentID)
-
 	case protocol.MessageTypeMetrics:
 		return h.handleMetricsMessage(ctx, agentID, data)
 
@@ -154,8 +151,13 @@ func (h *AgentHandler) newClient(agentID string, conn *websocket.Conn) *ws.Clien
 	}
 }
 
-func (h *AgentHandler) handleHeartbeatMessage(ctx context.Context, agentID string) error {
-	return h.agentService.UpdateAgentStatus(ctx, agentID, 1)
+func (h *AgentHandler) handleWebSocketPong(agentID string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := h.agentService.UpdateAgentStatus(ctx, agentID, 1); err != nil {
+		h.logger.Warn("failed to update agent status on pong", zap.String("agentID", agentID), zap.Error(err))
+	}
 }
 
 func (h *AgentHandler) handleMetricsMessage(ctx context.Context, agentID string, data json.RawMessage) error {

internal/protocol/message.go

@@ -50,7 +50,6 @@ const (
 	MessageTypeRegister    MessageType = "register"
 	MessageTypeRegisterAck MessageType = "register_ack"
 	MessageTypeRegisterErr MessageType = "register_error"
-	MessageTypeHeartbeat   MessageType = "heartbeat"
 	MessageTypeCommand     MessageType = "command"
 	MessageTypeCommandResp MessageType = "command_response"
 	MessageTypeUninstall   MessageType = "uninstall"

internal/websocket/manager.go

@@ -11,6 +11,12 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	serverPingInterval = 10 * time.Second
+	serverPongWait     = 30 * time.Second
+	serverWriteWait    = 10 * time.Second
+)
+
 // Client WebSocket客户端
 type Client struct {
 	ID         string          // 探针ID
@@ -31,11 +37,15 @@ type Manager struct {
 	mu         sync.RWMutex       // 读写锁
 	logger     *zap.Logger        // 日志
 	onMessage  MessageHandler     // 消息处理器
+	onPong     PongHandler        // Pong 处理器
 }
 
 // MessageHandler 消息处理器接口
 type MessageHandler func(ctx context.Context, probeID string, messageType string, data json.RawMessage) error
 
+// PongHandler Pong 处理器接口
+type PongHandler func(probeID string)
+
 // NewManager 创建新的WebSocket管理器
 func NewManager(logger *zap.Logger) *Manager {
 	return &Manager{
@@ -52,6 +62,11 @@ func (m *Manager) SetMessageHandler(handler MessageHandler) {
 	m.onMessage = handler
 }
 
+// SetPongHandler 设置 Pong 处理器
+func (m *Manager) SetPongHandler(handler PongHandler) {
+	m.onPong = handler
+}
+
 // Run 启动管理器
 func (m *Manager) Run(ctx context.Context) {
 	ticker := time.NewTicker(30 * time.Second)
@@ -202,10 +217,13 @@ func (c *Client) ReadPump(ctx context.Context) {
 		c.Conn.Close()
 	}()
 
-	c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
+	c.Conn.SetReadDeadline(time.Now().Add(serverPongWait))
 	c.Conn.SetPongHandler(func(string) error {
-		c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
+		c.Conn.SetReadDeadline(time.Now().Add(serverPongWait))
 		c.LastActive = time.Now()
+		if c.Manager.onPong != nil {
+			go c.Manager.onPong(c.ID)
+		}
 		return nil
 	})
 
@@ -238,7 +256,7 @@ func (c *Client) ReadPump(ctx context.Context) {
 
 // WritePump 向客户端写入消息
 func (c *Client) WritePump() {
-	ticker := time.NewTicker(30 * time.Second)
+	ticker := time.NewTicker(serverPingInterval)
 	defer func() {
 		ticker.Stop()
 		c.Conn.Close()
@@ -247,7 +265,7 @@ func (c *Client) WritePump() {
 	for {
 		select {
 		case message, ok := <-c.Send:
-			c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
+			c.Conn.SetWriteDeadline(time.Now().Add(serverWriteWait))
 			if !ok {
 				// 通道已关闭
 				c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
@@ -260,7 +278,7 @@ func (c *Client) WritePump() {
 			}
 
 		case <-ticker.C:
-			c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
+			c.Conn.SetWriteDeadline(time.Now().Add(serverWriteWait))
 			if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
 				return
 			}

pkg/agent/config/config.go

@@ -72,9 +72,6 @@ type CollectorConfig struct {
 	// 数据采集间隔（秒）
 	Interval int `yaml:"interval"`
 
-	// 心跳间隔（秒）
-	HeartbeatInterval int `yaml:"heartbeat_interval"`
-
 	// 网络采集包含的网卡列表（白名单，支持正则表达式）
 	// 如果配置了此项，则只采集匹配的网卡，忽略 NetworkExclude
 	// 例如: ["^eth0$", "^en0$", "^ens.*"]
@@ -120,8 +117,7 @@ func DefaultConfig() *Config {
 			LogCompress:   true,
 		},
 		Collector: CollectorConfig{
-			Interval:          5,
-			HeartbeatInterval: 30,
+			Interval: 5,
 		},
 		AutoUpdate: AutoUpdateConfig{
 			Enabled:       true,
@@ -205,10 +201,6 @@ func (c *Config) Validate() error {
 		c.Collector.Interval = 5
 	}
 
-	if c.Collector.HeartbeatInterval <= 0 {
-		c.Collector.HeartbeatInterval = 30
-	}
-
 	if c.AutoUpdate.Enabled {
 		if _, err := time.ParseDuration(c.AutoUpdate.CheckInterval); err != nil {
 			return fmt.Errorf("更新检查间隔格式错误: %w", err)
@@ -237,11 +229,6 @@ func (c *Config) GetCollectorInterval() time.Duration {
 	return time.Duration(c.Collector.Interval) * time.Second
 }
 
-// GetHeartbeatInterval 获取心跳间隔时长
-func (c *Config) GetHeartbeatInterval() time.Duration {
-	return time.Duration(c.Collector.HeartbeatInterval) * time.Second
-}
-
 // GetUpdateCheckInterval 获取更新检查间隔时长
 func (c *Config) GetUpdateCheckInterval() time.Duration {
 	duration, _ := time.ParseDuration(c.AutoUpdate.CheckInterval)

pkg/agent/service/agent.go

@@ -26,6 +26,12 @@ import (
 	"github.com/sourcegraph/conc"
 )
 
+const (
+	agentPingInterval = 10 * time.Second
+	agentPongWait     = 30 * time.Second
+	agentWriteWait    = 5 * time.Second
+)
+
 // safeConn 线程安全的 WebSocket 连接包装器
 type safeConn struct {
 	conn *websocket.Conn
@@ -164,14 +170,19 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 	// 创建线程安全的连接包装器
 	conn := &safeConn{conn: rawConn}
 
+	if err := rawConn.SetReadDeadline(time.Now().Add(agentPongWait)); err != nil {
+		return fmt.Errorf("设置读取超时失败: %w", err)
+	}
+	rawConn.SetPongHandler(func(string) error {
+		return rawConn.SetReadDeadline(time.Now().Add(agentPongWait))
+	})
+
 	// 设置 Ping 处理器，自动响应服务端的 Ping
 	rawConn.SetPingHandler(func(appData string) error {
-		// WriteControl 有内置锁，可以安全调用
-		err := rawConn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(time.Second))
-		if err == nil {
-			//slog.Info("收到 Ping，已发送 Pong")
+		if err := rawConn.SetReadDeadline(time.Now().Add(agentPongWait)); err != nil {
+			return err
 		}
-		return err
+		return rawConn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(agentWriteWait))
 	})
 
 	// 发送注册消息
@@ -204,11 +215,11 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 		}
 	})
 
-	// 启动心跳和数据发送
+	// 主动发送 Ping，快速检测断线
 	wg.Go(func() {
-		if err := a.heartbeatLoop(ctx, conn, done); err != nil {
+		if err := a.pingLoop(ctx, rawConn, done); err != nil {
 			select {
-			case errChan <- fmt.Errorf("心跳失败: %w", err):
+			case errChan <- fmt.Errorf("ping失败: %w", err):
 			default:
 			}
 		}
@@ -298,6 +309,24 @@ func (a *Agent) readLoop(conn *websocket.Conn, done chan struct{}) error {
 	}
 }
 
+func (a *Agent) pingLoop(ctx context.Context, conn *websocket.Conn, done chan struct{}) error {
+	ticker := time.NewTicker(agentPingInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(agentWriteWait)); err != nil {
+				return err
+			}
+		case <-done:
+			return nil
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
 // registerAgent 注册探针
 func (a *Agent) registerAgent(conn *safeConn) error {
 	// 加载或生成探针 ID
@@ -396,29 +425,6 @@ func (a *Agent) handleMonitorConfig(data json.RawMessage) {
 	}
 }
 
-// heartbeatLoop 心跳循环
-func (a *Agent) heartbeatLoop(ctx context.Context, conn *safeConn, done chan struct{}) error {
-	ticker := time.NewTicker(a.cfg.GetHeartbeatInterval())
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			if err := conn.WriteJSON(protocol.OutboundMessage{
-				Type: protocol.MessageTypeHeartbeat,
-				Data: struct{}{},
-			}); err != nil {
-				return fmt.Errorf("发送心跳失败: %w", err)
-			}
-			//slog.Info("心跳已发送")
-		case <-done:
-			return nil
-		case <-ctx.Done():
-			return nil
-		}
-	}
-}
-
 func (a *Agent) setActiveConn(conn *safeConn) {
 	a.connMu.Lock()
 	defer a.connMu.Unlock()

pkg/agent/service/service.go

@@ -274,8 +274,7 @@ func (m *ServiceManager) Run() error {
 
 	slog.Info("配置加载成功",
 		"server_endpoint", m.cfg.Server.Endpoint,
-		"collector_interval", m.cfg.GetCollectorInterval(),
-		"heartbeat_interval", m.cfg.GetHeartbeatInterval())
+		"collector_interval", m.cfg.GetCollectorInterval())
 
 	// 初始化系统配置（Linux ICMP 权限等）
 	configureICMP()