Golang网络服务心跳检测与维护方法

net.Dialer

net.ListenConfig

KeepAlive

map[string]time.Time

time.Ticker

// 示例：服务端心跳检测
type Client struct {
    ID        string
    Conn      net.Conn
    LastHeartbeat time.Time
    // ... 其他客户端信息
}

var clients = make(map[string]*Client)
var clientsMu sync.RWMutex

func handleConnection(conn net.Conn) {
    clientID := generateClientID() // 假设有生成ID的函数
    client := &Client{ID: clientID, Conn: conn, LastHeartbeat: time.Now()}
    clientsMu.Lock()
    clients[clientID] = client
    clientsMu.Unlock()

    go func() {
        defer func() {
            clientsMu.Lock()
            delete(clients, clientID)
            clientsMu.Unlock()
            conn.Close()
            log.Printf("Client %s disconnected.", clientID)
        }()

        // 启动一个goroutine接收客户端数据和心跳
        go clientReader(client)

        // 启动一个goroutine定期更新心跳时间（如果客户端发送心跳）
        // 或者这里是服务端主动发送心跳的逻辑
        for {
            select {
            case <-time.After(clientHeartbeatTimeout): // 客户端超时检测
                log.Printf("Client %s heartbeat timeout. Disconnecting.", clientID)
                return // 退出，触发defer关闭连接
            case <-client.Context.Done(): // 假设客户端有context来控制生命周期
                log.Printf("Client %s context cancelled. Disconnecting.", clientID)
                return
            }
        }
    }()
}

// clientReader 负责读取客户端数据，并更新LastHeartbeat
func clientReader(client *Client) {
    buf := make([]byte, 1024)
    for {
        n, err := client.Conn.Read(buf)
        if err != nil {
            // 处理错误，如io.EOF或网络错误
            return
        }
        // 假设心跳消息是特定的字符串或结构
        if string(buf[:n]) == "HEARTBEAT" {
            client.LastHeartbeat = time.Now() // 更新心跳时间
            log.Printf("Received heartbeat from client %s", client.ID)
        } else {
            // 处理业务数据
            log.Printf("Received data from client %s: %s", client.ID, string(buf[:n]))
        }
    }
}

time.Ticker

// 示例：客户端心跳发送与重连
func startClient(serverAddr string) {
    var conn net.Conn
    var err error
    reconnectInterval := 1 * time.Second

    for {
        if conn == nil {
            log.Printf("Connecting to %s...", serverAddr)
            conn, err = net.Dial("tcp", serverAddr)
            if err != nil {
                log.Printf("Failed to connect: %v. Retrying in %v...", err, reconnectInterval)
                time.Sleep(reconnectInterval)
                reconnectInterval = min(reconnectInterval*2, 60*time.Second) // 指数退避
                continue
            }
            log.Printf("Connected to %s", serverAddr)
            reconnectInterval = 1 * time.Second // 重置重连间隔
            go handleClientConnection(conn)
        }
        time.Sleep(1 * time.Second) // 简单等待，让handleClientConnection处理
    }
}

func handleClientConnection(conn net.Conn) {
    defer conn.Close()
    heartbeatTicker := time.NewTicker(5 * time.Second) // 每5秒发送一次心跳
    defer heartbeatTicker.Stop()

    // 启动一个goroutine接收服务端数据
    go func() {
        buf := make([]byte, 1024)
        for {
            n, err := conn.Read(buf)
            if err != nil {
                log.Printf("Read from server error: %v. Connection likely closed.", err)
                // 这里需要一种机制通知外层循环重新连接
                return
            }
            log.Printf("Received from server: %s", string(buf[:n]))
        }
    }()

    for range heartbeatTicker.C {
        _, err := conn.Write([]byte("HEARTBEAT"))
        if err != nil {
            log.Printf("Failed to send heartbeat: %v. Connection likely closed.", err)
            // 这里也需要通知外层循环重新连接
            return
        }
        log.Println("Sent heartbeat to server.")
    }
}

select

// 示例：使用select实现连接的读、写和心跳管理
func manageConnection(conn net.Conn, heartbeatInterval time.Duration) {
    readerCh := make(chan []byte)
    errorCh := make(chan error, 1) // 缓冲通道，避免发送阻塞

    go func() {
        buf := make([]byte, 1024)
        for {
            conn.SetReadDeadline(time.Now().Add(heartbeatInterval * 2)) // 读超时，用于检测对端是否发送数据或心跳
            n, err := conn.Read(buf)
            if err != nil {
                errorCh <- err
                return
            }
            readerCh <- append([]byte(nil), buf[:n]...) // 复制一份数据，避免并发问题
        }
    }()

    heartbeatTicker := time.NewTicker(heartbeatInterval)
    defer heartbeatTicker.Stop()

    for {
        select {
        case data := <-readerCh:
            // 处理接收到的数据，可能是业务数据，也可能是对端的心跳响应
            log.Printf("Received data: %s", string(data))
            // 如果是心跳响应，可以更新一个“最后活跃时间”
        case err := <-errorCh:
            log.Printf("Connection read error: %v", err)
            // 连接断开或超时，触发清理和重连逻辑
            return
        case <-heartbeatTicker.C:
            // 定期发送心跳
            _, err := conn.Write([]byte("PING"))
            if err != nil {
                log.Printf("Failed to send heartbeat: %v", err)
                // 发送失败，连接可能已断开
                return
            }
        // case <-ctx.Done(): // 如果有外部取消信号
        //     log.Println("Connection context cancelled.")
        //     return
        }
    }
}

net.Conn

SetReadDeadline

SetWriteDeadline

SetReadDeadline

Read

SetWriteDeadline

Write

// 结合deadline和心跳的例子
func readWithDeadline(conn net.Conn, timeout time.Duration) ([]byte, error) {
    conn.SetReadDeadline(time.Now().Add(timeout))
    buf := make([]byte, 1024)
    n, err := conn.Read(buf)
    if err != nil {
        // 判断是否是超时错误
        if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
            log.Println("Read timeout!")
        }
        return nil, err
    }
    return buf[:n], nil
}

Read

io.EOF

Read

connection reset by peer

io.EOF

// 简单的指数退避重连逻辑
func reconnectLoop(ctx context.Context, serverAddr string, handler func(net.Conn)) {
    var conn net.Conn
    var err error
    baseDelay := 1 * time.Second
    maxDelay := 60 * time.Second
    currentDelay := baseDelay

    for {
        select {
        case <-ctx.Done():
            log.Println("Reconnect loop cancelled.")
            if conn != nil {
                conn.Close()
            }
            return
        default:
            if conn == nil {
                log.Printf("Attempting to connect to %s (delay: %v)...", serverAddr, currentDelay)
                conn, err = net.Dial("tcp", serverAddr)
                if err != nil {
                    log.Printf("Connection failed: %v. Retrying.", err)
                    time.Sleep(currentDelay)
                    currentDelay = min(currentDelay*2, maxDelay)
                    continue
                }
                log.Printf("Successfully reconnected to %s.", serverAddr)
                currentDelay = baseDelay // 重置延迟
                go handler(conn) // 启动连接处理
            }
            time.Sleep(1 * time.Second) // 避免空循环CPU飙升
        }
    }
}

func min(a, b time.Duration) time.Duration {
    if a < b {
        return a
    }
    return b
}

defer conn.Close()

context.WithCancel