示例说明
每连接一个 goroutine 直接互发消息会随连接数平方级增长锁竞争。Hub 模式把状态变更收敛到单事件循环：注册、注销、广播三类事件全部经通道排队，处理天然串行无需细粒度锁。每客户端带缓冲的 send 通道把网络写入与广播解耦——写慢的客户端只是缓冲堆积，不会拖住广播循环。
技术要点
- 通道串行化消除状态竞争
- 房间映射支持多租户隔离
- 缓冲满即移除慢消费者
代码解读
run 的 select 三分支对应三类事件：注册加入房间、注销清理并关闭 send 通道、广播遍历房间内全部客户端非阻塞投递——default 分支意味着缓冲满的客户端被异步注销。serveWS 升级连接后启动读写两个协程，读协程刷新读超时兼当断线探测。
使用建议
读超时配 ping 心跳维持真实活跃检测。房间数量要治理防泄漏——空房间已即时删除。横向扩展时房间状态需外置到 Redis。
代码示例
package hub
import (
"log"
"net/http"
"sync"
"time"
"github.com/gorilla/websocket"
)
type Client struct{ conn *websocket.Conn; room string; send chan []byte }
type Hub struct {
mu sync.RWMutex
rooms map[string]map[*Client]struct{}
register chan *Client
unregister chan *Client
broadcast chan message
}
type message struct{ room string; data []byte }
func newHub() *Hub {
return &Hub{
rooms: map[string]map[*Client]struct{}{},
register: make(chan *Client),
unregister: make(chan *Client),
broadcast: make(chan message, 64),
}
}
func (h *Hub) run() {
for {
select {
case c := <-h.register:
h.mu.Lock()
if h.rooms[c.room] == nil {
h.rooms[c.room] = map[*Client]struct{}{}
}
h.rooms[c.room][c] = struct{}{}
h.mu.Unlock()
case c := <-h.unregister:
h.mu.Lock()
if room, ok := h.rooms[c.room]; ok {
if _, alive := room[c]; alive {
delete(room, c)
close(c.send)
}
if len(room) == 0 {
delete(h.rooms, c.room)
}
}
h.mu.Unlock()
case msg := <-h.broadcast:
h.mu.RLock()
for c := range h.rooms[msg.room] {
select {
case c.send <- msg.data:
default: // 缓冲满：掉队者被移除
go h.unregister <- c
}
}
h.mu.RUnlock()
}
}
}
var upgrader = websocket.Upgrader{ReadBufferSize: 1024, WriteBufferSize: 1024}
func (h *Hub) serveWS(w http.ResponseWriter, r *http.Request, room string) {
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Println("upgrade:", err)
return
}
client := &Client{conn: conn, room: room, send: make(chan []byte, 16)}
h.register <- client
go func() {
defer func() { h.unregister <- client }()
for {
conn.SetReadDeadline(time.Now().Add(90 * time.Second))
if _, _, err := conn.ReadMessage(); err != nil {
return
}
}
}()
go func() {
for data := range client.send {
if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
return
}
}
}()
}