Documentation
¶
Overview ¶
Package wsutil 提供通用 WebSocket 基础设施。
两个原语，都**不依赖任何 websocket 库**：
- Broadcaster：基于房间的广播器，支持多房间、自动清理空房间、容错广播。 通过 Conn 接口解耦。
- Outbox：单条连接的**有界出站管道**——保证「只有一个写者」，并把「慢 消费者」从发送方的主路径上摘掉。
用法（广播）：
b := wsutil.NewBroadcaster()
defer b.Close()
// 客户端加入房间
b.Register("room:123", conn)
// 广播事件
b.BroadcastEvent("room:123", wsutil.Event{Type: "progress", Data: data})
// 查询房间状态
count := b.GetRoomCount("room:123")
rooms := b.GetRooms()
用法（单连接的出站管道）：
// websocket 实现不许并发写，所以每条连接得有一个独家写者。
ob := wsutil.NewOutbox(func(msg []byte) error {
return conn.WriteMessage(websocket.TextMessage, msg)
}, wsutil.OutboxOptions{QueueSize: 64})
defer ob.Close()
ob.Send(frame) // 永不阻塞；队列满了返回 false，处置由调用方决定
Index ¶
- Constants
- type Broadcaster
- func (b *Broadcaster) Broadcast(room string, message []byte)
- func (b *Broadcaster) BroadcastEvent(room string, event Event)
- func (b *Broadcaster) Close()
- func (b *Broadcaster) GetRoomCount(room string) int
- func (b *Broadcaster) GetRooms() []string
- func (b *Broadcaster) GetTotalConnections() int
- func (b *Broadcaster) Register(room string, conn Conn)
- func (b *Broadcaster) Unregister(room string, conn Conn)
- type Conn
- type Event
- type Outbox
- type OutboxOptions
Constants ¶
const DefaultOutboxQueueSize = 64
DefaultOutboxQueueSize 是出站队列的默认长度。
64 帧够应付正常的突发（一次状态刷新往往连带发好几帧），又不至于让一个 慢客户端占住太多内存。
const (
// TextMessage 表示文本消息类型（与 websocket.TextMessage 一致）。
TextMessage = 1
)
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broadcaster ¶
type Broadcaster struct {
// contains filtered or unexported fields
}
Broadcaster 是基于房间的 WebSocket 广播器。 支持多房间、自动清理空房间、容错广播。
用法:
b := wsutil.NewBroadcaster()
defer b.Close()
// 客户端加入房间
b.Register("review:123", conn)
// 广播事件
b.BroadcastEvent("review:123", wsutil.Event{Type: "progress", Data: progress})
func (*Broadcaster) Broadcast ¶
func (b *Broadcaster) Broadcast(room string, message []byte)
Broadcast 向指定房间广播原始字节消息。
⚠️ **message 会被拷贝一份**（#78，与 Outbox.Send 同口径）。投递进 channel 之后、WriteMessage 真正执行之前有一段异步窗口；调用方若复用 buffer （buf = buf[:0] 再填、循环里 marshal 进同一块内存——高性能广播路径的 标准写法），在途消息会被改写成另一帧的内容：客户端收到串包/坏 JSON， 服务端无任何报错。一次拷贝换「广播出去的字节确定是 Broadcast 返回那一刻 的字节」。BroadcastEvent 每次 marshal 出新切片，天然安全。
func (*Broadcaster) BroadcastEvent ¶
func (b *Broadcaster) BroadcastEvent(room string, event Event)
BroadcastEvent 向指定房间广播事件（自动序列化为 JSON 并添加时间戳）。
func (*Broadcaster) GetRoomCount ¶
func (b *Broadcaster) GetRoomCount(room string) int
GetRoomCount 返回指定房间的连接数。
func (*Broadcaster) GetTotalConnections ¶
func (b *Broadcaster) GetTotalConnections() int
GetTotalConnections 返回所有房间的总连接数。
func (*Broadcaster) Register ¶
func (b *Broadcaster) Register(room string, conn Conn)
Register 将连接加入指定房间。
func (*Broadcaster) Unregister ¶
func (b *Broadcaster) Unregister(room string, conn Conn)
Unregister 将连接从指定房间移除。 同一连接可注册多个房间；物理连接仅在最后一个房间移除时关闭。
type Conn ¶
type Conn interface {
WriteMessage(messageType int, data []byte) error
Close() error
// SetWriteDeadline 设置写超时。它是**必选**方法，且必须真正生效——
// no-op 实现属违约：WriteMessage 可能永久阻塞，Broadcaster 对该连接
// 只能跳过后续写（见 handleBroadcast 的 #87 说明）。
SetWriteDeadline(t time.Time) error
}
Conn 是 WebSocket 连接的最小接口。 github.com/hertz-contrib/websocket.Conn 满足此接口。
type Event ¶
type Event struct {
Type string `json:"type"`
Timestamp time.Time `json:"timestamp"`
Data any `json:"data,omitempty"`
}
Event 是一个可序列化的事件，携带类型和时间戳。
type Outbox ¶ added in v0.7.0
type Outbox struct {
// contains filtered or unexported fields
}
Outbox 是单条连接的有界出站管道。
零值不可用，用 NewOutbox。
func NewOutbox ¶ added in v0.7.0
func NewOutbox(write func([]byte) error, opts OutboxOptions) *Outbox
NewOutbox 建一条出站管道并启动它的写协程。
write 是真正的写入动作（通常就是 conn.WriteMessage）。它只被写协程调用。
⚠️ **write 必须自带超时或可中断**：Close() 只关队列，打不断已经在跑的 一次 write。对端半开时，没有超时的 write 会让 Wait() 永远不返回 ⇒ 调用方 的关停路径挂死、goroutine 泄漏。传一个带写超时（SetWriteDeadline）的写入 动作，是调用方的责任。
func (*Outbox) Close ¶ added in v0.7.0
func (o *Outbox) Close()
Close 关闭这条管道。
幂等。关闭后写协程会**冲刷队列里剩余的帧**再退出——正在推的最后一帧不该 因为对端刚要走就被丢掉。
这个承诺没有例外（#117）：Close 持写锁置 closed、关 done，而 Send 的 「检查 + 入队」持读锁——Close 的临界区开始时，所有通过了检查的在途 Send 都已经把帧放进队列；之后不再有帧能入队。所以写协程冲刷到队列空时，队列 里不会有「迟到的帧」，Send(true) − Written ≡ Dropped 的记账公式把关闭 瞬间也闭合在内。
func (*Outbox) Closed ¶ added in v0.7.13
Closed 报告管道是否已关闭。
它拆开 Send 的两种 false（#105）：Send 返回 false 且 Closed() 为 true ⇒ 管道已关（那一帧没进队列，也不计入 Dropped/Rejected）；Closed() 为 false ⇒ 队列满（慢消费者，计 Rejected、按 Send 文档的示例处置）。文档给的示例把 false 一律当慢消费者断对端——管道已关时那样做是误伤，先用这个分清楚。
func (*Outbox) Done ¶ added in v0.7.0
func (o *Outbox) Done() <-chan struct{}
Done 在管道关闭后被关闭。
⚠️ **它不代表队列已冲刷** —— 它只表示「已关闭，不再接受新帧」。写协程 可能还在冲刷队列里剩下的帧。要确认「帧都写出去了」，必须等 Wait() 返回。
正确顺序是：**Close() → Wait() → 关底层连接**。
（这一段曾经写着「等 Done 再关底层连接」—— 而那恰恰是会截断待发帧的 顺序：Done 在 Close() 里就关闭了，那时一帧都还没冲刷。）
func (*Outbox) Dropped ¶ added in v0.7.0
Dropped 是「Send 返回 true（已进队列）、但没有写出去」的帧数（#123 拆分口径）。
两种情形都算（#105 闭合口径）：正在写但写失败的那一帧、以及写失败后随队列 一起消失的帧（含 Close 后冲刷阶段的写失败）。Send 返回 true 的每一帧都必然 闭合进 Written 或 Dropped 之一——「Send(true) 次数 − Written」恒等于 Dropped。
关闭瞬间同样闭合（#117）：Send 的检查与入队是同一临界区，不存在「检查 通过了、帧却落在关闭之后」的交错——那类帧曾经两头都不记，公式差值恒增 1 且无法归因。
⚠ 旧口径把「队列满拒收」也混在本计数里，恒等式在任何发生过背压的进程上 必假——消费方照「Send(true)−Written」检测写失败丢帧会拿到全然失真的数字， 无法区分慢消费与写失败（#123）。拒收帧从未进过队列、Send 返回 false，不属 于「丢了已收下的帧」，现由 Outbox.Rejected 单独计数。
它非零说明连接已经**写不动了**（写失败即关闭管道）——那是**该去看的东西**： 检查写入端的超时与对端状态。慢消费（队列满）看 Rejected 而不是这里。
func (*Outbox) Rejected ¶ added in v0.8.0
Rejected 是**队列满拒收**的帧数（#123）：Send 因队列满返回 false 的次数。
这些帧从未进过队列、调用方收到了 false 并被明示「没收下」，因此不计入 Dropped——旧的混记口径使「Send(true) − Written ≡ Dropped」恒等式失真， 且让慢消费与写失败无法区分。管道已关时的拒收**不计**（帧同样没进队列， 但那属于关闭语义，Closed() 可区分）。
它非零说明有客户端读得比服务端推得快——慢消费是该去看的东西：要么客户端 该自己慢点读，要么推送的频率该降，要么按 Send 文档的示例处置对端。
func (*Outbox) Send ¶ added in v0.7.0
Send 把一帧放进队列。
它**永不阻塞**：队列满时返回 false，返回 true 只说明「进了队列」，不说明 「已经写到网络上」。
**它不替调用方做处置**——满了之后是断开这条连接、断开对端、丢掉这一帧， 还是退出本端，取决于场景。所以「满了」只报告、不动作：
if !ob.Send(frame) {
closePeer(p) // 慢消费者：断开这条连接
}
关闭之后再 Send 返回 false，不会 panic。两种 false 可区分（#105）： 管道已关（ob.Closed() 为 true——帧根本没进队列，不算丢帧）与队列满 （Closed() 为 false——慢消费者，拒收计 Rejected（#123），按文档示例处置对端）。
「是否已关闭」的检查与入队是**同一个临界区**（#117）：返回 true 的帧 必然已经在队列里，写协程的冲刷承诺（见 Close）对它成立——不存在 「检查通过了、帧却落在关闭之后」的交错。返回 false 且 Closed() 为 true 的帧则必然没进队列。
⚠️ **帧会被拷贝一份**。写协程是**稍后**才写它的，而调用方很可能复用 同一个 buffer（buf = buf[:0] 后重新填）——不拷贝的话，上线的是调用方改过 之后的字节，症状是「客户端收到坏 JSON」，而服务端这边什么错都不报。 拷贝换来的是一次分配；相比「坏字节难查」，这笔买卖划算。
type OutboxOptions ¶ added in v0.7.0
type OutboxOptions struct {
// QueueSize 出站队列长度。零即 DefaultOutboxQueueSize。
QueueSize int
}
OutboxOptions 是 NewOutbox 的可选项。
Source Files