深入理解Go语言TCP连接：优雅处理客户端断开与写入错误

package main

import (
    "bufio"
    "fmt"
    "net"
    "os"
    "sync" // 引入sync包用于互斥锁
)

// Connection 结构体封装了TCP连接和其故障状态
type Connection struct {
    Conn      net.Conn
    IsFaulted bool
    mu        sync.Mutex // 用于保护IsFaulted字段的并发访问
}

// StartWritingToNetwork 负责向客户端写入数据
// 它从msgStack通道接收消息，并尝试写入。如果写入失败，则标记连接为故障，
// 将消息放回msgStack（以便后续重发），并通过errChannel通知错误。
func StartWritingToNetwork(connWrap *Connection, errChannel chan<- error, msgStack <-chan string) {
    for {
        msg := <-msgStack // 从消息栈中取出消息

        connWrap.mu.Lock() // 锁定，检查连接状态
        if connWrap.IsFaulted {
            connWrap.mu.Unlock()
            // 连接已故障，将消息放回栈，并退出写入goroutine
            // 注意：这里简单地放回，实际生产环境可能需要更复杂的重发策略或死信队列
            select {
            case msgStack <- msg: // 尝试放回，避免阻塞
            default:
                // 如果通道已满，则丢弃消息，或者记录日志
                fmt.Printf("Warning: msgStack full, dropping message: %s", msg)
            }
            return
        }
        connWrap.mu.Unlock() // 解锁

        _, err := connWrap.Conn.Write([]byte(msg))
        if err != nil {
            fmt.Printf("failed sending a message to network: %v\n", err)

            connWrap.mu.Lock() // 锁定，更新连接状态
            connWrap.IsFaulted = true
            connWrap.mu.Unlock()

            // 将未发送成功的消息放回栈，以便其他连接或重连后处理
            select {
            case msgStack <- msg: // 尝试放回，避免阻塞
            default:
                fmt.Printf("Warning: msgStack full during error, dropping message: %s", msg)
            }

            errChannel <- err // 通过错误通道通知主循环连接故障
            return            // 写入goroutine退出
        } else {
            fmt.Printf("msg sent: %s", msg)
        }
    }
}

// StartReadingFromNetwork 负责从客户端读取数据
// 它持续读取数据，如果读取失败（包括EOF），则标记连接为故障，
// 并通过errChannel通知错误。
func StartReadingFromNetwork(connWrap *Connection, errChannel chan<- error) {
    networkReader := bufio.NewReader(connWrap.Conn)
    for {
        connWrap.mu.Lock() // 锁定，检查连接状态
        if connWrap.IsFaulted {
            connWrap.mu.Unlock()
            return // 连接已故障，退出读取goroutine
        }
        connWrap.mu.Unlock() // 解锁

        line, err := networkReader.ReadString('\n')
        if err != nil {
            fmt.Printf("failed reading from network: %v\n", err)

            connWrap.mu.Lock() // 锁定，更新连接状态
            connWrap.IsFaulted = true
            connWrap.mu.Unlock()

            errChannel <- err // 通过错误通道通知主循环连接故障
            return            // 读取goroutine退出
        } else {
            fmt.Printf("Received from client: %s", line) // 打印收到的消息
        }
    }
}

// AcceptConnections 负责接受新的客户端连接，并为每个连接启动读写goroutine
func AcceptConnections(listener net.Listener, consoleMsgQueue chan string) {
    errChannel := make(chan error, 1) // 使用带缓冲的错误通道，避免阻塞

    for {
        conn, err := listener.Accept()
        if err != nil {
            fmt.Printf("Error accepting connection: %v\n", err)
            continue // 继续尝试接受下一个连接
        }

        fmt.Printf("client connected from %s\n", conn.RemoteAddr())

        // 创建一个Connection封装，并初始化为非故障状态
        connWrap := &Connection{Conn: conn, IsFaulted: false}

        // 为每个新连接启动独立的读写goroutine
        go StartReadingFromNetwork(connWrap, errChannel)
        go StartWritingToNetwork(connWrap, errChannel, consoleMsgQueue)

        // 阻塞等待当前连接的错误通知。一旦收到错误，表示该连接已故障。
        // 在实际应用中，这里可能需要更复杂的连接管理逻辑，例如记录日志、清理资源等。
        <-errChannel
        fmt.Printf("Connection from %s faulted. Closing connection.\n", conn.RemoteAddr())
        conn.Close() // 关闭故障连接，释放资源
    }
}

// ReadConsole 负责从控制台读取用户输入，并将消息发送到网络消息队列
func ReadConsole(networkMsgQueue chan<- string) {
    consoleReader := bufio.NewReader(os.Stdin)
    for {
        fmt.Print("Enter message to send: ")
        line, err := consoleReader.ReadString('\n')
        if err != nil {
            panic(fmt.Sprintf("Error reading from console: %v", err))
        }
        networkMsgQueue <- line // 将控制台输入发送到网络消息队列
    }
}

func main() {
    listener, err := net.Listen("tcp", "localhost:6666")
    if err != nil {
        panic(fmt.Sprintf("Error listening: %v", err))
    }
    defer listener.Close()

    println("listening on " + listener.Addr().String())

    // consoleToNetwork 是一个缓冲通道，用于存放待发送的消息
    // 缓冲大小可以根据实际需求调整，防止发送过快导致阻塞
    consoleToNetwork := make(chan string, 100)

    // 启动接受连接的goroutine
    go AcceptConnections(listener, consoleToNetwork)

    // 主goroutine负责从控制台读取输入并放入消息队列
    ReadConsole(consoleToNetwork)
}