README
¶
stats（统计域）
bitmap 在线状态 + 指标 + 健康——谁在线的 O(1) 位查询，在线规模的全量视图
这个域干什么
stats 回答两个问题：「某个用户现在在线吗」（点查询）和「现在有多少人在线、
分布在哪」（面查询），同时承载 Hub 的指标与健康度上报
管：在线状态（bitmap 点查 / 批查 / 全量游标分页）、指标聚合、健康检查 （HubHealthInfo）、在线状态与 Redis 的同步
不管：连接注册表（connection 管「进程内这条连接」，本域管「集群视角谁在线」）
核心概念
| 概念 | 说明 |
|---|---|
| bitmap 在线 | uid → offset（hash 分桶）→ SETBIT/GETBIT；1M 在线 ≈ 125 KB（REFACTORING_DESIGN §五.4） |
| Pipeline 批查 | BatchIsUserOnline：HGET offset + GETBIT 打包 Pipeline，一次往返 |
| uid→offset LRU | 进程内缓存，容量联动 WS_ONLINE_MAX_CACHED_UIDS；部分 shard 清理防命中率雪崩 |
| 游标分页 | GetAllOnlineUsers 用 SCAN/HSCAN 游标，不阻塞主循环（P8） |
| 本地兜底 | OnlineStore 未启用时回退本地注册表查询 |
Host 端口 |
ports.go 声明，hub 实现——后端异常时降级本地 |
怎么用
package main
import (
"context"
"github.com/kamalyes/go-wsc/stats"
)
mgr := stats.NewManager(deps)
// 点查询
online, err := mgr.IsUserOnline(ctx, "user-1001")
// 批量查询（Pipeline 合并，一次往返）
results, err := mgr.BatchIsUserOnline(ctx, "user-1001", "user-1002", "user-1003")
// 全量在线（游标分页，绝不 ZRANGE 0 -1）
err := mgr.GetAllOnlineUsers(ctx, func(batch []string) error {
// 每批处理
return nil
})
// 健康与指标
health := mgr.Health()
端口与依赖
- 依赖 spi 契约
OnlineStore（一等 Redis 实现：go-wsc-redis-adapter 的 bitmap） - 依赖 connection 注册表做本地兜底
性能与陷阱
- P7：uid→offset 频繁查映射 → 1M+ 次 RTT；必须 LRU 缓存 + 容量与在线规模联动
- P8：
ZRANGE 0 -1全量扫描 → Redis 阻塞 + 内存尖峰；游标分页是唯一正解 - P13：bitmap 冷启动缺 offset → 首查慢；先查一次映射再入 LRU
- 热点 key：bitmap 桶
bm:<app>:<ns>已 hash 分桶 64，防单 slot 热点
后端绑定
- Redis（一等 OnlineStore）：bitmap 在线状态 + uid_map 分桶映射
- 不启用时：本地注册表兜底，只答本节点在线（单节点语义完整）
Documentation
¶
Index ¶
- type Host
- type Manager
- func (m *Manager) CheckUserOnline(ctx context.Context, userID string) bool
- func (m *Manager) GetAllOnlineUserIDs() ([]string, error)
- func (m *Manager) GetHubHealth() *models.HubHealthInfo
- func (m *Manager) GetOnlineUserCount() (int64, error)
- func (m *Manager) GetOnlineUsersByNode(nodeID string) ([]string, error)
- func (m *Manager) GetStats() *models.HubStats
- func (m *Manager) GetUptime() int64
- func (m *Manager) IsUserOnline(ctx context.Context, userID string) (bool, error)
- func (m *Manager) LogClientConnection(client *models.Client)
- func (m *Manager) ShouldTrackUserStats(userType models.UserType) bool
- func (m *Manager) SyncClientStats()
- func (m *Manager) SyncOnlineStatus(client *models.Client)
- func (m *Manager) SyncOnlineStatusToRedis() error
- func (m *Manager) TrackConnectionError(connectionID string, userType models.UserType, err error)
- func (m *Manager) TrackHeartbeatStats(client *models.Client)
- func (m *Manager) TrackReceiverMessageStats(connectionID string, receiverType models.UserType, dataSize int)
- func (m *Manager) TrackSenderMessageStats(connectionID string, senderType models.UserType)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Host ¶
type Host interface {
// Context 返回 Hub 生命周期上下文（仅用于日志关联；追踪路径优先用
// client.Context 以保留连接级 trace_id）
Context() context.Context
// GetLogger 日志器
GetLogger() spi.Logger
// GetNodeID 当前节点 ID（统计上报的维度之一）
GetNodeID() string
// GetStartTime Hub 启动时间（运行时长统计）
GetStartTime() time.Time
// IsStarted Hub 是否已启动（健康状态）
IsStarted() bool
// GetShardedRegistry 连接注册表（读取实时连接数）
GetShardedRegistry() *connection.ShardedRegistry
// GetStatsRepo 节点统计仓储；未注入时返回 nil，追踪退化为 no-op
GetStatsRepo() spi.HubStats
// GetOnlineStatusRepo 在线状态仓储；未注入时返回 nil
GetOnlineStatusRepo() spi.OnlineStore
// GetConnectionQualityRepository 连接质量仓储；未注入时返回 nil
GetConnectionQualityRepository() spi.ConnectionQualityStore
// GetConnectionRecordRepo 连接记录仓储；未注入时返回 nil
GetConnectionRecordRepo() spi.ConnectionStore
// GetMessageStatsBatcher 消息统计批量聚合器；未启用时返回 nil
GetMessageStatsBatcher() *batcher.MessageStatsBatcher
// GetHeartbeatBatcher 心跳统计批量聚合器；未启用时返回 nil
GetHeartbeatBatcher() *batcher.HeartbeatStatsUpdater
// GetErrorStatsBatcher 连接错误统计批量聚合器；未启用时返回 nil
GetErrorStatsBatcher() *batcher.ErrorStatsBatcher
}
Host 统计域所需的 Hub 能力面。
追踪方法（Track*）是「事件 → 统计」的单向漏斗：由连接生命周期与消息 投递路径调用，本包据此判断是否采样、并把增量交给批量聚合器。
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager 统计域管理器
func (*Manager) CheckUserOnline ¶
CheckUserOnline 在线检查（支持分布式，按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入；本地与 Redis 查询均按 appID+namespace 信封过滤， 避免同名 userID 跨 app/ns 误判在线导致不必要的跨节点路由
性能热路径：调 IsUserOnline（bitmap 启用时 HGET uid_map → GETBIT 单次 Lua 往返， 未启用或无路由信封时回退 ZCount/全量过滤，永远最终一致）。 不调 GetUserNodes（ZRANGE + GET ×N + N 次解压，热路径不可接受）； 跨节点路由取节点列表由 distributed.go::checkAndRouteToNode 专用路径负责
func (*Manager) GetAllOnlineUserIDs ¶
GetAllOnlineUserIDs 获取所有在线用户 ID 列表 有存储后端时查询全量（含其他节点）；无后端时返回本地注册表视图 （分片注册表是 WS + SSE 客户端的超集）。
func (*Manager) GetHubHealth ¶
func (m *Manager) GetHubHealth() *models.HubHealthInfo
GetHubHealth 获取健康状态快照（使用 shardedRegistry 原子计数器）
func (*Manager) GetOnlineUserCount ¶
GetOnlineUserCount 获取在线用户总数
func (*Manager) GetOnlineUsersByNode ¶
GetOnlineUsersByNode 获取指定节点的在线用户 ID
func (*Manager) IsUserOnline ¶
IsUserOnline 检查用户是否在线（按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入，调用方不需传 appID/namespace 参数 本地检查走 shardedRegistry.HasUser，跨节点检查走 onlineStatusRepo.IsUserOnline （bitmap 启用时 HGET→GETBIT 单次往返，未启用时回退 ZCount/全量过滤，与原 IsUserOnline 等价）
func (*Manager) LogClientConnection ¶
LogClientConnection 记录客户端连接日志（单行 KV，高频路径）
func (*Manager) ShouldTrackUserStats ¶
ShouldTrackUserStats 判断是否应该追踪用户统计（排除系统、机器人、观察者）
func (*Manager) SyncClientStats ¶
func (m *Manager) SyncClientStats()
SyncClientStats 同步客户端统计信息到 Redis（内部获取最新连接数）
func (*Manager) SyncOnlineStatus ¶
SyncOnlineStatus 同步在线状态到指定存储
func (*Manager) SyncOnlineStatusToRedis ¶
SyncOnlineStatusToRedis 同步当前所有在线用户到存储（Hub 启动时或定期对账）
func (*Manager) TrackConnectionError ¶
TrackConnectionError 追踪连接错误 优化：使用批量聚合器，避免每次错误都启动 goroutine 直写数据库 断连风暴 3000 连接 = 3000 次 Submit + 攒批后 1 次事务（与消息统计同模式）
func (*Manager) TrackHeartbeatStats ¶
TrackHeartbeatStats 追踪心跳和 Ping 统计 优化：使用批量聚合器，避免每次心跳都启动 goroutine 写数据库 心跳时间戳由 batcher flush 写 connect 表，Ping 统计写 quality 表
落库节流：每连接 throttle 间隔内仅落一次（Client.StatsFlushDue）， CRDB 心跳批量 UPDATE 的 range 写热点随节流系数等比下降； last_pong_at 全链路只写不读（dashboard 同步但零查询）已停写
Source Files