Go 实现（二）：一个读一个写，读写分离与并发安全

gorilla/websocket 的并发契约只有两条：一个连接最多一个 goroutine 在读，最多一个 goroutine 在写（WriteControl 与 Close 例外，它们可以与其他方法并发）。业务里"服务端主动推消息"和"用户发消息"天然是两个来源，如果各自直接调 WriteMessage，就会撞上 panic: concurrent write to websocket connection。所以长连接网关的标准结构是：每个连接两个常驻协程——读协程（readPump）与写协程（writePump），所有出站消息先进入一个带缓冲的 channel。下面接着第 03 篇的同一个文件改造 wsHandler。

为什么用"写协程 + channel"而不是加锁

方案	优点	问题
全局锁保护写	改动最小	慢客户端持锁时拖住所有广播者，锁与写超时混在一起难定位
每个连接一把锁	故障隔离到单连接	仍会阻塞消息生产方，比如 Hub 的广播协程
单写协程 + 有缓冲 channel	生产方永不阻塞写、天然保序、队列长度可观测	需要处理队列满与关闭顺序

连接对象与幂等关闭

const (
	writeWait      = 10 * time.Second // 单次写超时
	pongWait       = 60 * time.Second // 读超时（配合 ping/pong，见第 06 篇）
	pingPeriod     = (pongWait * 9) / 10
	maxMessageSize = 32 << 10 // 32 KiB
	sendQueueSize  = 64       // 每个连接的出站队列长度
)

type Client struct {
	uid  string
	conn *websocket.Conn
	send chan []byte
	done chan struct{}
	once sync.Once
}

// Close 幂等：读协程退出、写协程出错、业务踢人、进程退出都可能触发
func (c *Client) Close() {
	c.once.Do(func() {
		close(c.done) // 通知读写两个协程退出
		_ = c.conn.Close()
	})
}

// Send 非阻塞投递：返回 false 说明队列满或连接已结束
func (c *Client) Send(msg []byte) bool {
	select {
	case c.send <- msg:
		return true
	case <-c.done:
		return false
	default:
		return false // 丢弃还是断开，是第 10 篇背压章节的取舍
	}
}

send 只由连接的所有者关闭（写协程或 Hub 的注销流程），绝不由投递方关闭；done 用来广播"这条连接结束了"，避免"向已关闭 channel 发送"这类 panic。

读协程：只负责读，不做重活

func (c *Client) readPump(onMessage func(msg []byte)) {
	defer c.Close() // 读协程退出即代表连接生命周期结束

	c.conn.SetReadLimit(maxMessageSize)
	_ = c.conn.SetReadDeadline(time.Now().Add(pongWait)) // 读超时，pong 续命见第 06 篇
	c.conn.SetPongHandler(func(string) error { return c.conn.SetReadDeadline(time.Now().Add(pongWait)) })

	for {
		_, msg, err := c.conn.ReadMessage()
		if err != nil {
			var ce *websocket.CloseError
			if errors.As(err, &ce) { // 关闭码要落日志，排障全靠它
				log.Printf("uid=%s 关闭 code=%d reason=%q", c.uid, ce.Code, ce.Text)
			}
			return
		}
		onMessage(msg) // 耗时逻辑丢给 worker/消息队列，别阻塞读
	}
}

读协程是"连接是否还活着"的唯一真相来源，它退出时必须触发注销（Hub 层处理，见第 05 篇）。反过来说，读协程里不要做重活：解析大 JSON、写库、调 RPC 都会让读停顿，TCP 接收缓冲被填满后对端立刻感到卡顿。

写协程：所有出站字节（含心跳）的唯一出口

func (c *Client) writePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.Close()
	}()

	for {
		select {
		case <-c.done:
			// 体面关闭：尽力发出 close 帧，失败也无所谓
			_ = c.conn.WriteControl(websocket.CloseMessage,
				websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
				time.Now().Add(writeWait))
			return
		case msg := <-c.send:
			_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
				return // 写失败即认为连接不可用
			}
		case <-ticker.C:
			_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

组装：把两个协程挂起来

// 升级成功后：先注册，再启动写协程与读协程
c := &Client{uid: uid, conn: conn, send: make(chan []byte, sendQueueSize), done: make(chan struct{})}
h.Register(c)
go c.writePump()
go func() {
	defer h.Unregister(c) // 读退出 → 注销 → 回收，这条链不能断
	c.readPump(func(msg []byte) { h.Broadcast(msg) })
}()

并发安全的几条纪律

• 所有出站数据都必须经过 send channel，包括系统通知与"被踢下线"的 close 帧；广播复用同一个 []byte 是安全的（写出是同步完成的），但投递后不能再改这块内存，要异步持有就先 bytes.Clone(msg)；
• SetReadDeadline 只由读协程调用，SetWriteDeadline 只由写协程调用，避免两个协程互相覆盖对方的超时；关闭连接也一样，统一走 Client.Close()，不要在别处调 c.conn.Close()；
• 用 go run -race . 或 go test -race ./... 跑一遍竞态检测，比线上偶发 panic 便宜得多。

常见坑与排查

• concurrent write to websocket connection panic：存在绕过 send 的直写路径，通常是"系统通知/被踢下线"直接调了 WriteMessage；在 readPump 里直接回写也会踩同一个坑。
• 协程泄漏或消息静默丢失：writePump 只 select send 不 select done，读协程退出后它仍阻塞在 channel 上（第 17 篇）；队列满时 Send 返回 false 被忽略，丢弃还是断开要显式决策并打点。

小结：每个连接两个常驻协程，读协程负责探活与收消息，写协程独占所有出站字节，生产方只做非阻塞 channel 投递，关闭动作收敛到幂等的 Close()。这套结构把并发问题从"到处加锁"变成了"一条单向数据流"，后面接 Hub 广播、心跳、背压、优雅重启，都只是在它上面加策略。