Go WebSocket：并发处理与多客户端消息广播实践

package main

import (
    "fmt"
    "io"
    "log"
    "net/http"

    "golang.org/x/net/websocket" // 使用标准导入路径
)

// connAddCh 用于向中央管理器发送新连接
var connAddCh = make(chan *websocket.Conn)

// connRemoveCh 用于向中央管理器发送断开的连接
var connRemoveCh = make(chan *websocket.Conn)

// msgBroadcastCh 用于向中央管理器发送待广播的消息
var msgBroadcastCh = make(chan []byte)

// connectionManager Goroutine 负责管理所有活跃的WebSocket连接并进行消息广播
func connectionManager() {
    // 使用 map 来存储活跃连接，键为 *websocket.Conn，值为空结构体{}作为占位符
    // map[*websocket.Conn]struct{} 是一种高效的集合实现
    activeConnections := make(map[*websocket.Conn]struct{})

    for {
        select {
        case newConn := <-connAddCh:
            // 添加新连接
            activeConnections[newConn] = struct{}{}
            log.Printf("新客户端连接。当前活跃连接数：%d", len(activeConnections))

        case disconnectedConn := <-connRemoveCh:
            // 移除断开的连接
            if _, ok := activeConnections[disconnectedConn]; ok {
                delete(activeConnections, disconnectedConn)
                disconnectedConn.Close() // 确保连接关闭
                log.Printf("客户端断开连接。当前活跃连接数：%d", len(activeConnections))
            }

        case msg := <-msgBroadcastCh:
            // 广播消息给所有活跃连接
            messageWithNewline := append(msg, '\n') // 添加换行符以模拟聊天消息

            // 收集需要移除的连接，避免在迭代map时修改map
            var connectionsToRemove []*websocket.Conn
            for conn := range activeConnections {
                // 尝试向客户端写入消息
                if _, err := conn.Write(messageWithNewline); err != nil {
                    // 如果写入失败，通常意味着客户端已断开或网络异常
                    log.Printf("向客户端 %s 发送消息失败: %v。标记为待移除。", conn.RemoteAddr(), err)
                    connectionsToRemove = append(connectionsToRemove, conn)
                }
            }
            // 移除所有发送失败的连接
            for _, conn := range connectionsToRemove {
                if _, ok := activeConnections[conn]; ok { // 再次检查以防并发问题
                    delete(activeConnections, conn)
                    conn.Close() // 确保连接关闭
                    log.Printf("移除因发送失败而断开的客户端 %s。当前活跃连接数：%d", conn.RemoteAddr(), len(activeConnections))
                }
            }
        }
    }
}

// EchoServer 处理单个WebSocket连接的读写
func EchoServer(ws *websocket.Conn) {
    // 连接建立时，将此连接发送到添加通道
    connAddCh <- ws

    // 当此处理函数退出时（连接关闭），将此连接发送到移除通道
    defer func() {
        connRemoveCh <- ws
        log.Printf("客户端 %s 的处理函数退出。", ws.RemoteAddr())
    }()

    log.Printf("开始处理客户端: %s", ws.RemoteAddr())

    buff := make([]byte, 512) // 读取缓冲区，可根据消息大小调整
    for {
        n, err := ws.Read(buff)
        if err != nil {
            if err == io.EOF {
                log.Printf("客户端 %s 正常断开。", ws.RemoteAddr())
            } else {
                log.Printf("从客户端 %s 读取错误: %v", ws.RemoteAddr(), err)
            }
            break // 读取出错或EOF时退出循环
        }
        // 将读取到的消息发送到广播通道