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 向指定房间广播原始字节消息。
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 设置写超时（可选实现）。
// 如果 Conn 不实现此方法，Broadcaster 会跳过写超时设置。
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 关闭这条管道。
幂等。关闭后写协程会**冲刷队列里剩余的帧**再退出——正在推的最后一帧不该 因为对端刚要走就被丢掉。
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 是**没有写出去**的帧数。
两种情形都算：队列满时拒收的帧，以及写失败后随队列一起消失的帧。
它非零说明有客户端读得比服务端推得快——那是**该去看的东西**：要么客户端 该自己慢点读，要么推送的频率该降。
func (*Outbox) Send ¶ added in v0.7.0
Send 把一帧放进队列。
它**永不阻塞**：队列满时返回 false，返回 true 只说明「进了队列」，不说明 「已经写到网络上」。
**它不替调用方做处置**——满了之后是断开这条连接、断开对端、丢掉这一帧， 还是退出本端，取决于场景。所以「满了」只报告、不动作：
if !ob.Send(frame) {
closePeer(p) // 慢消费者：断开这条连接
}
关闭之后再 Send 返回 false，不会 panic。
⚠️ **帧会被拷贝一份**。写协程是**稍后**才写它的，而调用方很可能复用 同一个 buffer（buf = buf[:0] 后重新填）——不拷贝的话，上线的是调用方改过 之后的字节，症状是「客户端收到坏 JSON」，而服务端这边什么错都不报。 拷贝换来的是一次分配；相比「坏字节难查」，这笔买卖划算。
type OutboxOptions ¶ added in v0.7.0
type OutboxOptions struct {
// QueueSize 出站队列长度。零即 DefaultOutboxQueueSize。
QueueSize int
}
OutboxOptions 是 NewOutbox 的可选项。
Source Files