Go语言WebSocket服务器：实现多客户端消息广播与连接管理

package main

import (
    "fmt"
    "io"
    "log"
    "net/http"
    "sync" // For a more robust connection storage example, though not strictly needed for the single-goroutine approach
    "golang.org/x/net/websocket"
)

// 定义两个通道，用于协程间通信
// c: 用于传递新的WebSocket连接到管理协程
// c2: 用于传递客户端发送的消息到管理协程
var c = make(chan *websocket.Conn, 5) // 缓冲区大小可根据实际并发量调整
var c2 = make(chan []byte, 5)

// activeConnections 存储所有活跃的WebSocket连接。
// 在本例中，它由一个单独的goroutine独占管理，因此无需额外的锁。
// 如果需要在其他goroutine中直接访问，则需要sync.Mutex。
type ConnectionStore struct {
    mu    sync.RWMutex // 读写锁，如果ConnectionStore的方法被多个goroutine调用时需要
    conns map[*websocket.Conn]struct{} // 使用map[conn]struct{}实现set
}

func NewConnectionStore() *ConnectionStore {
    return &ConnectionStore{
        conns: make(map[*websocket.Conn]struct{}),
    }
}

// Add 添加一个连接
func (cs *ConnectionStore) Add(conn *websocket.Conn) {
    cs.mu.Lock()
    defer cs.mu.Unlock()
    cs.conns[conn] = struct{}{}
    log.Printf("New connection added: %s. Total active connections: %d", conn.RemoteAddr(), len(cs.conns))
}

// Remove 移除一个连接
func (cs *ConnectionStore) Remove(conn *websocket.Conn) {
    cs.mu.Lock()
    defer cs.mu.Unlock()
    delete(cs.conns, conn)
    log.Printf("Connection removed: %s. Total active connections: %d", conn.RemoteAddr(), len(cs.conns))
}

// GetConnections 获取所有连接的副本，用于迭代
func (cs *ConnectionStore) GetConnections() []*websocket.Conn {
    cs.mu.RLock()
    defer cs.mu.RUnlock()
    var connections []*websocket.Conn
    for conn := range cs.conns {
        connections = append(connections, conn)
    }
    return connections
}


// EchoServer 是每个WebSocket连接的处理器
func EchoServer(ws *websocket.Conn) {
    defer func() {
        log.Printf("Handler for %s closing.", ws.RemoteAddr())
        ws.Close() // 确保连接关闭
    }()

    // 将新连接发送到新连接通道，由管理协程处理
    c <- ws

    buff := make([]byte, 512) // 读取缓冲区
    for {
        // 从客户端读取消息
        size, err := ws.Read(buff)
        if err != nil {
            if err == io.EOF {
                log.Printf("Client %s disconnected.", ws.RemoteAddr())
            } else {
                log.Printf("Read error from %s: %v", ws.RemoteAddr(), err)
            }
            break // 读取错误或客户端断开，退出循环
        }

        // 将读取到的消息发送到消息通道，由管理协程广播
        message := make([]byte, size) // 创建一个精确大小的切片
        copy(message, buff[:size])
        c2 <- message
        log.Printf("Received message from %s: %s", ws.RemoteAddr(), string(message))
    }
}

func main() {
    // 启动一个独立的协程，作为连接管理器和消息广播器
    go func() {
        // somekindofstorage 存储活跃连接
        // 在此示例中，我们使用一个简单的map，因为所有操作都在这个goroutine中完成
        // 如果ConnectionStore的方法被其他goroutine调用，则需要其内部的锁。
        connectionStore := NewConnectionStore()

        for {
            select {
            case newC := <-c: // 接收新的WebSocket连接
                connectionStore.Add(newC)
            case msg := <-c2: // 接收需要广播的消息
                log.Printf("Broadcasting message: %s", string(msg))
                // 遍历所有活跃连接，发送消息
                for _, conn := range connectionStore.GetConnections() {
                    if _, err := conn.Write(msg); err != nil {
                        // 写入失败通常意味着客户端已断开或网络问题
                        log.Printf("Failed to write to %s, removing connection: %v", conn.RemoteAddr(), err)
                        connectionStore.Remove(conn) // 从活跃连接中移除
                    }
                }
            }
        }
    }()

    // 注册WebSocket处理器
    http.Handle("/echo", websocket.Handler(EchoServer))

    // 启动HTTP服务器
    port := ":12345"
    log.Printf("WebSocket server starting on %s", port)
    if err := http.ListenAndServe(port, nil); err != nil {
        log.Fatalf("Server failed to start: %v", err)
    }
}

// 如果ConnectionStore的方法被多个goroutine调用，则需要如下保护
type ConnectionStore struct {
    mu    sync.RWMutex // 读写锁
    conns map[*websocket.Conn]struct{}
}

func (cs *ConnectionStore) Add(conn *websocket.Conn) {
    cs.mu.Lock() // 写操作加锁
    defer cs.mu.Unlock()
    cs.conns[conn] = struct{}{}
}

func (cs *ConnectionStore) Remove(conn *websocket.Conn) {
    cs.mu.Lock() // 写操作加锁
    defer cs.mu.Unlock()
    delete(cs.conns, conn)
}

func (cs *ConnectionStore) GetConnections() []*websocket.Conn {
    cs.mu.RLock() // 读操作加读锁
    defer cs.mu.RUnlock()
    var connections []*websocket.Conn
    for conn := range cs.conns {
        connections = append(connections, conn)
    }
    return connections
}