连接管理器与 Hub：注册、注销、广播与房间

连接管理器（connection hub）是长连接网关的心脏：维护"谁在线、在哪个房间、消息该发给谁"，还要在十万连接规模下保持 O(1) 的注册与注销。常见写法是"一个大 map + 一把全局锁，谁广播谁加锁遍历"，连接数一上来就会因锁竞争与慢客户端雪崩；更稳的做法是事件循环（event loop）：注册表的读写全部收敛到 Hub 自己的 goroutine 串行执行，map 干脆不需要锁。职责只有三块——连接表（谁在线）、会话索引（uid → 该用户的全部连接，支撑多端在线）、房间表（room → 连接集合）；鉴权在握手期完成（第 08 篇），跨节点路由交给外部索引（ims-route），Hub 都不管。

结构：单协程串行化所有改动

type roomOp struct {
	client *Client // 房间消息广播时为 nil
	room   string
	data   []byte // 仅广播时有值
	join   bool
}

// Hub 用单协程串行处理所有改动：channel 带缓冲（如 256），map 只初始化一次，内部读写无需加锁
type Hub struct {
	register   chan *Client
	unregister chan *Client
	broadcast  chan []byte
	roomOps    chan roomOp // 加入 / 退出 / 房间广播都走这里

	clients map[*Client]struct{}
	rooms   map[string]map[*Client]struct{}
	byUID   map[string]map[*Client]struct{}
	online  atomic.Int64 // 在线连接总数，读侧无锁
}

事件循环：所有改动的唯一入口

func (h *Hub) Run() {
	for {
		select {
		case c := <-h.register:
			h.clients[c] = struct{}{}
			h.online.Add(1)
			h.addUID(c)
		case c := <-h.unregister:
			if _, ok := h.clients[c]; !ok {
				break // 已注销过，保证幂等
			}
			delete(h.clients, c)
			h.online.Add(-1)
			h.delUID(c)
			for room := range c.rooms {
				h.detach(c, room)
			}
			c.Close() // 触发 readPump/writePump 退出
		case op := <-h.roomOps:
			switch {
			case op.data != nil: // 房间广播
				for c := range h.rooms[op.room] {
					c.Send(op.data) // 非阻塞投递，慢客户端不拖累别人
				}
			case op.join:
				h.attach(op.client, op.room)
			default:
				h.detach(op.client, op.room)
			}
		case msg := <-h.broadcast:
			for c := range h.clients {
				c.Send(msg)
			}
		}
	}
}

func (h *Hub) addUID(c *Client) {
	if c.uid == "" {
		return // 匿名连接不进会话索引
	}
	if h.byUID[c.uid] == nil {
		h.byUID[c.uid] = make(map[*Client]struct{}, 2)
	}
	h.byUID[c.uid][c] = struct{}{}
}

func (h *Hub) delUID(c *Client) {
	if set := h.byUID[c.uid]; set != nil {
		delete(set, c)
		if len(set) == 0 {
			delete(h.byUID, c.uid) // 清空空集合，避免索引无限膨胀
		}
	}
}

break 在 select 的 case 里退出的是整个 select，等于"跳过本次处理的剩余逻辑"，这是幂等注销最省事的写法。进程退出时遍历 clients 逐个 Close() 即可（优雅发布见第 15 篇）。

房间与多端索引

Client.rooms 只在 Hub 协程里被读写，所以不需要锁——这是事件循环模型最大的好处：并发安全问题变成了代码纪律问题。

func (h *Hub) attach(c *Client, room string) {
	set := h.rooms[room]
	if set == nil {
		set = make(map[*Client]struct{})
		h.rooms[room] = set
	}
	set[c] = struct{}{}
	if c.rooms == nil {
		c.rooms = make(map[string]struct{}, 4)
	}
	c.rooms[room] = struct{}{}
}
func (h *Hub) detach(c *Client, room string) {
	if set := h.rooms[room]; set != nil {
		delete(set, c)
		if len(set) == 0 {
			delete(h.rooms, room) // 空房即回收
		}
	}
	delete(c.rooms, room)
}

对内 API：调用方只发事件

对外只暴露六个不阻塞的方法：Register、Unregister、Broadcast、BroadcastRoom、Join、Leave，实现都是一行——把事件投进对应的 channel（例如 func (h *Hub) Broadcast(msg []byte) { h.broadcast <- msg }）。需要返回值的查询（"某用户当前有几端在线"）也走"请求—应答"channel：调用方带一个 resp chan int 进来，Hub 在 select 里回填结果。不要在 Hub 协程之外读这些 map，那等于把刚消除的锁问题换成了数据竞争。

实现方式	读性能	写 / 广播	适用规模
全局 sync.RWMutex + map	一般（读锁竞争）	遍历时持锁，易被慢客户端拖住	连接数 < 数千
单协程事件循环（本章）	查询绕行协程，略慢	串行、可预测、易加策略	十万连接级网关
分片锁 / sync.Map	高	广播仍需遍历全部	想并发写又不想收敛到单点

广播只有三种取舍：遍历中直写 socket（一个慢客户端拖住整轮）、非阻塞投递到出站队列（本章做法）、以及把消息落到共享存储再让各连接自己拉。完整策略见第 10 篇背压与第 16 篇推送；Hub 只保证一件事——投递不阻塞。

常见坑与排查

• 在 Hub 协程里做阻塞操作、或向自己的 channel 发送：直写 socket、查库、调 RPC 会让整个 Hub 停摆，表现为全站消息延迟飙升；事件循环里出现 h.register <- c 则直接死锁，所有对外方法必须在 Hub 协程之外调用。
• 注销不幂等、空集合不回收、广播缓冲被改写：重复注销会重复扣减在线数甚至关闭已关闭的连接；rooms / byUID 残留空 map 会让内存持续上涨；广播投递的是切片引用，投递后再覆写这块内存会造成脏数据。

小结：Hub 的骨架是"事件循环 + 索引表 + 非阻塞投递"——注册、注销、房间变更、广播全部作为事件进入单协程串行处理，连接对象只保留出站队列与只读元数据。这样既避免了全局锁竞争，也把并发安全变成了"谁在哪个协程里碰哪些字段"的纪律；在线数与路由一旦要跨节点，就该交给外部索引，而不是继续往单机内存里堆。