Golang TCP服务端近乎并发接收消息时遭遇EOF异常

问题场景 #

我开发了一个Golang TCP服务端用于和设备通信，约定流程如下：

• 设备先发hello消息，服务端回复hello
• 设备发送数据，完成后发送commit请求确认数据是否接收

当前遇到的问题是：服务端无法收到commit请求，日志显示设备发送数据与commit请求的间隔仅0.1秒（近乎并发），但服务端在收到数据后立即出现EOF错误：

Device IP: 13.56.127.98:65145, Timestamp: 2023-08-07T21:23:40.893412845Z
Error reading header: EOF

服务端核心代码如下：

package servers

import (
    "bufio"
    "bytes"
    "encoding/binary"
    "fmt"
    "math/rand"
    "net"
    "time"
)

func Server() {
    PORT := ":8081"
    l, err := net.Listen("tcp", PORT)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer l.Close()
    rand.Seed(time.Now().Unix())

    for {
        c, err := l.Accept()
        if err != nil {
            fmt.Println(err)
            return
        }

        // Set TCP keep-alive settings
        if tcpConn, ok := c.(*net.TCPConn); ok {
            // Enable TCP keep-alive
            tcpConn.SetKeepAlive(true)

            // Set the time period for sending TCP keep-alive probes
            keepAlivePeriod := 60 * time.Hour // Adjust this value as needed
            tcpConn.SetKeepAlivePeriod(keepAlivePeriod)
        }
        go func() {
            defer c.Close()
            handleConnection(c)
        }()
    }
}

func handleConnection(c net.Conn) {
    defer c.Close()

    data := make([]byte, 4096)

    for {
        //log device ip
        fmt.Printf("Device IP: %s, Timestamp: %s \n", c.RemoteAddr().String(), time.Now().Format(time.RFC3339Nano))
        // Read the communication header from the connection
        //_, err := c.Read(data)
        _, err := bufio.NewReader(c).Read(data)
        if err != nil {
            fmt.Println("   Error reading header:", err)
            return
        }

        // Check the sync bytes
        if data[0] != 0x02 || data[1] != 0x55 {
            fmt.Println("   Invalid sync bytes")
            //log actual bytes received
            fmt.Printf("    Received: %02X %02X %02X %02X %02X\n", data[0], data[1], data[2], data[3], data[4])
            return
        }

        // Get the message type from the header
        messageType := data[2]

        // get payload length
        payloadLengthBytes := data[3:5]
        payloadLength := binary.LittleEndian.Uint16(payloadLengthBytes)

        // Handle the different message types
        switch messageType {
        case 0:
            fmt.Println("   Received Hello message:")
            sendHelloResponse(c)
        case 4:
            // Send data records
            fmt.Println("   Received Send data records message:")
        case 5:
            // Commit request
            fmt.Println("   Received Commit request message:")
        // ... more cases follow
        }
    }
}

我推测问题出在循环读取数据逻辑，且单条消息仅约500字节，缓冲区大小足够。

问题根源分析 #

1. 重复创建bufio.Reader：在handleConnection的循环内每次调用bufio.NewReader(c)，会导致上一次读取后留在缓冲区的字节被丢弃。TCP是流式协议，设备发送的data消息和commit请求可能被一次性读到第一个bufio.Reader的缓冲区中，处理完data后，下一次循环新建的Reader无法读取到缓冲区里的commit请求，最终因数据读完/设备关闭连接触发EOF。
2. 忽略Read返回的实际字节数：代码完全忽略了Read的返回值n，可能导致处理时用上一次读取的残留脏数据，干扰消息判断。
3. 未保证读取完整消息：当前只做了一次Read操作，若消息被拆分为多个TCP段，会导致读取不完整，后续逻辑无法正确识别消息类型。

解决方案 #

1. 复用bufio.Reader #

在handleConnection开头仅创建一次bufio.Reader，复用整个连接周期：

func handleConnection(c net.Conn) {
    defer c.Close()
    // 仅创建一次Reader，复用至连接关闭
    reader := bufio.NewReader(c)
    data := make([]byte, 4096)

    for {
        fmt.Printf("Device IP: %s, Timestamp: %s \n", c.RemoteAddr().String(), time.Now().Format(time.RFC3339Nano))
        // 使用复用的reader读取数据
        n, err := reader.Read(data)
        if err != nil {
            fmt.Println("   Error reading header:", err)
            return
        }
        // 校验头部完整性，避免处理不完整数据
        if n < 5 {
            fmt.Println("   Received incomplete header")
            continue
        }
        // 后续处理仅针对已读取的n字节数据，避免脏数据干扰
        // ... 原头部校验、消息类型判断逻辑保持不变，但操作data[:n]范围

2. 读取完整消息体 #

根据头部中的payloadLength读取完整的消息内容，避免因TCP分段导致的读取不完整：

// 获取payload长度后，读取完整消息体
payloadLength := binary.LittleEndian.Uint16(data[3:5])
totalMsgLen := 5 + int(payloadLength) // 头部5字节+payload长度

if totalMsgLen > len(data) {
    // 缓冲区不足时，创建临时数组存储完整消息
    fullMsg := make([]byte, totalMsgLen)
    copy(fullMsg[:n], data[:n])
    remaining := totalMsgLen - n
    offset := n
    for remaining > 0 {
        readN, err := reader.Read(fullMsg[offset:])
        if err != nil {
            fmt.Println("   Error reading payload:", err)
            return
        }
        remaining -= readN
        offset += readN
    }
    // 使用fullMsg进行后续处理
} else {
    // 缓冲区足够，读取剩余的payload内容
    remaining := totalMsgLen - n
    offset := n
    for remaining > 0 {
        readN, err := reader.Read(data[offset:])
        if err != nil {
            fmt.Println("   Error reading payload:", err)
            return
        }
        remaining -= readN
        offset += readN
    }
}

3. 排查设备侧连接状态 #

EOF通常意味着设备主动关闭了连接，可通过抓包工具验证设备是否确实发送了commit请求，以及是否在发送后立即断开连接。

内容的提问来源于stack exchange，提问作者LUKER