connection

package
v0.12.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 1, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

README

connection（连接域）

百万连接的注册 / 查找 / 心跳 / 容量——64 分片注册表是整个库的地基

这个域干什么

connection 管理连接的全生命周期数据结构：一个连接进来后它在哪注册、怎么按 userID 找到它、多设备怎么算一个人、心跳超时怎么管、慢消费者怎么隔离

管：分片注册表（注册/注销/按 userID 查找）、心跳管理（HeartbeatManager： 时间轮 O(1) 超时 + SSE 兜底扫描）、连接生命周期（LifecycleManager：多端登录 治理 / 踢出断链 / 精简移除）、连接记录（RecordManager：快照构造 / 异步落库 / 停机批量终态）、连接容量（capacity）、控制通道（control lane）、 慢消费者识别与降级

不管：消息语义（messaging）、接入握手（transport）、在线状态存储（stats + Redis）

核心概念

概念 说明
ShardedRegistry 64 个 map[userID][]*Client 分片，FNV-1a hash(userID) % 64 定位
分片数 max(CPU核数, 64)：锁竞争比单 map 降 64 倍（REFACTORING_DESIGN §五.1）
ClientMatchesEnvelope appID / namespace / groupID 信封匹配，广播过滤的判据
RegistryCapacity 容量上限 + 拒绝策略（满载时新连接直接拒绝）
HeartbeatManager 心跳管理器：前置回调 → 时间轮续期 → Redis 续期入队 → 回调链 → 统计
LifecycleManager 生命周期管理器：多端登录策略（禁多端踢旧 / 上限踢最旧）、踢出断链（先通知后注销）、shutdown 精简移除
MultiLoginPolicy 多端登录策略快照：AllowMultiLogin + MaxConnectionsPerUser，按 appID+namespace 信封隔离
RecordManager 连接记录管理器：Client → ConnectionRecord 快照构造、异步 Upsert、停机批量标记断开（1001）
control lane 必达通道：心跳/ACK/控制帧绕过普通队列，慢消费者也不丢
慢消费者 连续 N 次写超阈值的连接标记降级，防积压拖垮全 Hub
多设备 同一 userID 多条连接：注册表按 []*Client 存，user/device 双维度计数

怎么用

package main

import (
    "github.com/kamalyes/go-wsc/connection"
)

reg := connection.NewShardedRegistry(connection.RegistryCapacity{
    MaxConnections: 1_000_000,
})

// 注册 / 注销
reg.Register(client)      // 连接建立
reg.Unregister(client)   // 连接关闭（引用计数，最后一条才清 user 身份）

// 查找
clients := reg.GetClients("user-1001")   // 该用户全部活跃连接（多设备）

// 遍历（广播用）：shard 级并行
reg.ForEachClientParallel(workerNum, func(c *connection.Client) {
    // writev 合批写……
})

端口与依赖

  • 定义 interfaces.go 消费者端口（EvictHook / HeartbeatHost / LifecycleHost / RecordHost），由 hub 实现
  • 连接记录仓储经 RecordHost 端口动态读取（spi.ConnectionStore），支持运行期注入；其余零外部依赖：纯内存数据结构（含时间轮），不 import 任何存储驱动

性能与陷阱

  • 量级：100 万连接 ≈ 200 MB 注册表索引（不含帧缓冲，全量估算见 REFACTORING_DESIGN §五.1）
  • P1：分片数不随核数——4 核机器跑 64 分片反而慢，用 max(CPU核数, 64)
  • P2：慢消费者积压——必须 control_lane 优先 + 降级，否则一个慢连接拖垮全 Hub
  • P3：伪共享——每 shard 独立 struct 填充到 cache line 边界（50 万连接以上再看）

后端绑定

纯内存在线状态的持久化视角（谁在线）归 stats 域 + Redis OnlineStore； 本域只管「进程内这条连接」

Documentation

Index

Constants

View Source 
const DefaultClientSendChanCapacity = 256

DefaultClientSendChanCapacity 默认客户端 SendChan 缓冲区容量（当配置不存在时使用）

Variables

This section is empty.

Functions

func ClassifyCloseError 

func ClassifyCloseError(err error) (closeCode int, isNormal bool)

ClassifyCloseError 分类关闭错误

func ClientMatchesEnvelope 

func ClientMatchesEnvelope(client *models.Client, msgAppID, msgNamespace string, msgGroupIDs []string) bool

ClientMatchesEnvelope 判断某个 client 是否应该接收携带指定路由信封的消息

投递匹配规则：

  1. appId 维度（最上层隔离）：严格相等，无空值兼容（appID 无全局语义，入口层必归一化为 DefaultAppID）
  2. namespace 维度：msgNamespace 非空时严格相等；msgNamespace=="" 表示全局广播，跳过 ns 过滤匹配所有

设计说明：

  • appID 严格匹配：appID 是应用隔离维度，不存在"全局跨 app 广播"语义，空值在入口层归一化为 DefaultAppID
  • namespace 空值兼容：全局广播（跨所有命名空间的系统通知）是合法业务需求， msgNamespace=="" 表示全局广播，匹配所有 ns 的 client（与 Broadcast 全局语义一致）

⚠️ 不做 msgGroupIDs vs client.GroupID 匹配（与 ForEachUserClientFiltered 对称设计）：

  • msg.GroupIDs 存放的是"业务群组ID"（如 g-broadcast），而 client.GroupID 是"连接级系统组" （如 __default_gp__），两者完全两个维度，强行匹配导致客户端全部被过滤（delivered=0）
  • 群组成员过滤已由 groupRepo.GetMembers + memberSet 完成（broadcastToUserIDs 路径）， 全局广播（broadcastToFiltered/ForEachClientFiltered）无需也不应做系统组匹配
  • msgGroupIDs 参数仅作签名兼容，不参与匹配

本函数为热路径（每次投递遍历 client 都调用）：零分配、内联友好、短路优先

func CopyClientsFromMap 

func CopyClientsFromMap(clientMap map[string]*models.Client) []*models.Client

CopyClientsFromMap 从客户端映射中复制客户端列表 用于避免在遍历时map被修改导致的数据竞争

func GetClientIP 

func GetClientIP(client *models.Client) string

GetClientIP 从客户端获取IP地址

func IsControlMessage 

func IsControlMessage(mt models.MessageType) bool

IsControlMessage 判断是否为控制类消息（控制通道 + 兜底降级的路由依据）

func IsDisconnectMessage 

func IsDisconnectMessage(mt models.MessageType) bool

IsDisconnectMessage 判断是否为断链类控制消息（CtrlCh 满时降级为直接断链）

Types

type CapacityManager 

type CapacityManager struct {
	// contains filtered or unexported fields
}

CapacityManager 按 UserType 配置通道容量并初始化客户端通道四件套 （DoneCh/PongCh/CtrlCh/SendChan）；channel 与 Client 同生命周期，随连接分配、断连后由 GC 回收

func NewCapacityManager 

func NewCapacityManager(cfg *wscconfig.ClientCapacity, logger spi.Logger) *CapacityManager

NewCapacityManager 创建容量管理器 cfg 为 nil 时使用 wscconfig.DefaultClientCapacity()；logger 为 nil 时使用默认日志器

func (*CapacityManager) CapacityFor 

func (cm *CapacityManager) CapacityFor(client *models.Client) int

CapacityFor 根据客户端 UserType 获取 SendChan 缓冲区容量（配置不存在项为 0 时由上层兜底）

func (*CapacityManager) InitClientSendChan 

func (cm *CapacityManager) InitClientSendChan(client *models.Client)

InitClientSendChan 初始化客户端的通道四件套（按 UserType 容量分配 SendChan） 在客户端注册时调用，确保 SendChan 使用正确的缓冲区大小

type ControlLane 

type ControlLane struct {
	// contains filtered or unexported fields
}

ControlLane 控制通道投递器（必达级消息的独立 lane）

func NewControlLane 

func NewControlLane(logger spi.Logger) *ControlLane

NewControlLane 创建控制通道投递器（logger 为 nil 时使用默认日志器）

func (*ControlLane) SendControl 

func (l *ControlLane) SendControl(client *models.Client, data []byte, msg *models.HubMessage) bool

SendControl 向客户端控制通道发送已序列化的必达级消息

降级链（逐级兜底，控制消息尽力可达）：

  1. CtrlCh 投递（首选：独立 lane，不受数据洪峰影响）
  2. CtrlCh 未初始化（手工构造/SSE 早期客户端）→ 回退数据 lane TrySend （无优先保障但可达——比直接断链后客户端不知原因好）
  3. 通道满且断链类（KickOut/ForceOffline/Terminate）：直接 Conn.Close() （断链本就是目的，通知帧是锦上添花）
  4. 通道满且非断链类：返回 false，调用方走各自兜底（ACK 有重试+超时转离线）

func (*ControlLane) SendControlMessage 

func (l *ControlLane) SendControlMessage(client *models.Client, msg *models.HubMessage) bool

SendControlMessage 向客户端控制通道发送控制类消息（内部序列化 + 控制通道投递一体化）

用于 KickOut/ForceOffline 等场景：消息经 Marshal 后走 CtrlCh，绕开业务 SendChan 洪峰

type DisconnectionSubmitter 

type DisconnectionSubmitter interface {
	// Submit 非阻塞提交断连终态条目（队列满时返回 false，
	// 与记录池可丢弃语义一致，仅影响审计统计精度）
	Submit(entry *models.DisconnectionEntry) bool
}

DisconnectionSubmitter 断连终态攒批提交端口（消费者侧端口：Hub 实现）

正常注销路径的断连终态经攒批合并落库（CASE WHEN 单 SQL），替代逐条 UPDATE + 每断一连一个超时 goroutine 直打 DB 的写放大路径

type EvictHook 

type EvictHook interface {
	OnSlowConsumerEvicted(client *models.Client, consecutive int, ratio float64)
}

EvictHook 驱逐回调端口（消费者侧端口：Hub 实现，扫描器不感知驱逐细节）

实现方职责（见 hub 编排层）：

  • 慢消费者驱逐埋点（overload 域指标）
  • KickOut 控制消息通知客户端驱逐理由
  • Unregister 断链（断链不丢消息）

type HeartbeatHost 

type HeartbeatHost interface {
	// Context 返回 Hub 生命周期上下文（仅用于日志关联）
	Context() context.Context
	// GetLogger 日志器
	GetLogger() spi.Logger
	// Unregister 异步注销客户端（心跳超时与 SSE 兜底触发断链）
	Unregister(client *models.Client)
	// TrackHeartbeatStats 心跳统计追踪（stats 域漏斗，不阻塞主流程）
	TrackHeartbeatStats(client *models.Client)
	// EnqueueHeartbeatRenew 心跳 Redis 在线索引续期入队（满则丢弃，下次心跳补投）
	EnqueueHeartbeatRenew(client *models.Client)

	// GetBeforeHeartbeatCallback 心跳前置回调（返回 false 跳过后续心跳处理）
	GetBeforeHeartbeatCallback() func(client *models.Client) bool
	// GetHeartbeatReportCallback 心跳上报回调（业务侧按心跳周期感知活跃度）
	GetHeartbeatReportCallback() func(client *models.Client)
	// GetAfterHeartbeatCallback 心跳后置回调
	GetAfterHeartbeatCallback() func(client *models.Client)
	// GetHeartbeatTimeoutCallback 心跳超时回调（超时注销时触发）
	GetHeartbeatTimeoutCallback() func(clientID string, userID string, lastHeartbeat time.Time)
}

HeartbeatHost 心跳子域所需的 Hub 能力面

心跳链路（前置拦截 → 时间轮续期 → Redis 续期入队 → 回调链 → 统计追踪） 的外部依赖经本端口注入，由 hub 编排层实现；回调 getter 用结构类型声明， 避免 hub ⇄ connection 循环 import（与 group.Host 回调端口同款约定）， 未注册时返回 nil，调用方跳过（不可假定非空）

type HeartbeatManager 

type HeartbeatManager struct {
	// contains filtered or unexported fields
}

HeartbeatManager 心跳管理器（连接域）

持有分片时间轮，管理 WebSocket 客户端的 O(1) 心跳超时；SSE 客户端 不发送 PING，无法通过时间轮 Refresh，由 ScanSSETimeouts 周期扫描 LastSeen 判断活跃度

func NewHeartbeatManager 

func NewHeartbeatManager(host HeartbeatHost, registry *ShardedRegistry, clientTimeout time.Duration, timerOptions ...syncx.TimerOption) *HeartbeatManager

NewHeartbeatManager 构造心跳管理器

构造期初始化时间轮（不依赖 Run），确保 Schedule/Refresh/Cancel 在 任何 goroutine 启动前可用；timerOptions 透传时间轮分片配置

func (*HeartbeatManager) CancelTimeout 

func (m *HeartbeatManager) CancelTimeout(clientID string)

CancelTimeout 注销连接时取消时间轮上的心跳超时任务

func (*HeartbeatManager) Handle 

func (m *HeartbeatManager) Handle(client *models.Client)

Handle 处理一次心跳（协议级 PING 与应用层心跳共用同一保活入口） 流程：前置回调 → 续期 → Redis 异步续期入队 → 上报回调 → 后置回调 → 统计

日志策略：成功路径单条 Debug（生产 Info 级静默，排查时开 Debug 可观测心跳节奏； 千万级下逐条 Info 是日志风暴，心跳频次经 TrackHeartbeatStats 聚合统计）

func (*HeartbeatManager) ScanSSETimeouts 

func (m *HeartbeatManager) ScanSSETimeouts()

ScanSSETimeouts 扫描 SSE 客户端心跳超时（兜底机制） WebSocket 客户端由时间轮 O(1) 管理，此处仅扫描 SSE 客户端 SSE 客户端不发送 PING，无法通过时间轮 Refresh，需定期扫描 LastSeen 判断活跃度

⚠️ 死锁防御：遍历持有 shard 读锁，先收集超时客户端到本地 slice（mutex 保护 并发 append），遍历结束后在锁外统一调用 Unregister

func (*HeartbeatManager) ScheduleTimeout 

func (m *HeartbeatManager) ScheduleTimeout(client *models.Client)

ScheduleTimeout 注册连接时在时间轮上调度心跳超时任务 仅用于 WebSocket 客户端；SSE 客户端由 ScanSSETimeouts 扫描兜底

func (*HeartbeatManager) Stop 

func (m *HeartbeatManager) Stop()

Stop 停止时间轮（不再触发超时注销；停机路径调用）

type LifecycleHost 

type LifecycleHost interface {
	// GetLogger 日志器
	GetLogger() spi.Logger
	// IsShuttingDown 编排层是否正在关闭（决定断链是否先发 1001 GoingAway 控制帧）
	IsShuttingDown() bool
	// Unregister 异步注销客户端（踢出路径断链）
	Unregister(client *models.Client)
	// SendToClient 经消息域向客户端投递（强制下线通知等控制类直达消息）
	SendToClient(ctx context.Context, client *models.Client, msg *models.HubMessage)
}

LifecycleHost 连接生命周期子域所需的 Hub 能力面

多端登录治理与踢出断链的外部依赖经本端口注入，由 hub 编排层实现

type LifecycleManager 

type LifecycleManager struct {
	// contains filtered or unexported fields
}

LifecycleManager 连接生命周期管理器（连接域）

覆盖注册后的准入治理（多端登录策略）与断链执行（踢出通知、 精简移除、半注册清理、通道/连接关闭原语）

func NewLifecycleManager 

func NewLifecycleManager(host LifecycleHost, registry *ShardedRegistry, heartbeat *HeartbeatManager, policy MultiLoginPolicy) *LifecycleManager

NewLifecycleManager 构造连接生命周期管理器

func (*LifecycleManager) CleanupHalfRegistered 

func (m *LifecycleManager) CleanupHalfRegistered(client *models.Client)

CleanupHalfRegistered 清理半注册连接（注册编排 panic 兜底） 半注册状态：连接已加入注册表 + 心跳超时任务已调度，但注册流程中断 幂等安全：各清理步骤对"未执行到"的步骤均为无操作

func (*LifecycleManager) CloseChannel 

func (m *LifecycleManager) CloseChannel(client *models.Client)

CloseChannel 关闭客户端发送通道 用 DoneCh 通知写循环退出，数据通道（SendChan/SSEMessageCh）永不 close：

  1. 消除 TrySend 的 chansend 与本函数 closechan 的数据竞态
  2. 不跨连接复用：竞态窗口内 racing sender 仍可能写入残留消息，复用会跨连接串消息； channel 与 Client 同生命周期，断连后随 Client 整体交由 GC 回收
  3. 不置 nil SendChan，避免与写循环 select 读产生数据竞争

func (*LifecycleManager) CloseConnection 

func (m *LifecycleManager) CloseConnection(client *models.Client)

CloseConnection 关闭 WebSocket 连接 Hub 关闭（如 K8s 滚动更新）时先发送 1001 GoingAway 控制帧， 让客户端识别为服务端主动离开并触发重连，而不是收到 1006 异常断开

func (*LifecycleManager) EnforceMultiLoginPolicy 

func (m *LifecycleManager) EnforceMultiLoginPolicy(newClient *models.Client)

EnforceMultiLoginPolicy 统一处理多端登录策略 根据策略决定是否允许多端登录、是否限制连接数：

  • AllowMultiLogin=false：踢掉该用户全部旧连接（同信封快速检查，无旧连接零开销）
  • MaxConnectionsPerUser>0：达到上限时踢掉最不活跃（最旧心跳）的旧连接

调用时新连接已入注册表（AddClient 前置），故踢人时排除新连接自身； 同 clientID 覆盖已由注销路径的指针一致性校验保护。 多端登录策略按 appID+namespace 信封隔离：不同应用/命名空间的连接互不影响 （app-A 的连接数不挤占 app-B 的配额）

func (*LifecycleManager) KickClient 

func (m *LifecycleManager) KickClient(client *models.Client, reason models.DisconnectReason, message string)

KickClient 踢掉客户端并发送强制下线通知 通知先于断开写入客户端发送通道，写循环异步发出；随后注销连接

func (*LifecycleManager) KickUser 

func (m *LifecycleManager) KickUser(ctx context.Context, userID string, reason string, sendNotification bool, notificationMsg string) *models.KickUserResult

KickUser 统一踢出用户全部连接（唯一踢人实现，按 ctx 路由信封 appID+namespace 隔离）

收集维度：按 ctx 路由信封过滤（与 P2P 投递同一隔离语义）——同名 userID 跨 app/namespace 多端在线时，仅踢出信封内的连接，不同应用/租户互不误踢； gRPC/distributed 路径经路由信封恢复/注入后传入，本地路径继承调用方信封

幂等语义：用户已无连接（收集数 0）即"已离线"目标达成，结果 KickedConnections=0 不视为失败；sendNotification=true 时先向全部连接写入 KickOut 通知再注销 （Guaranteed 级控制消息，断链前投递，notificationMsg 为通知文案）

ctx 由调用方传入（grpc/distributed 路径已恢复 trace_id），实现全链路追踪

func (*LifecycleManager) RemoveUnsafe 

func (m *LifecycleManager) RemoveUnsafe(client *models.Client)

RemoveUnsafe 从注册表移除客户端（shutdown 路径专用精简清理） 仅做：注册表移除 + 时间轮取消 + 指针一致性校验 + 关闭通道与连接； 不触发回调与逐条记录落库（由停机批量清理统一处理， 避免大量串行写 Redis/DB 导致 shutdown 超时）

type MultiLoginPolicy 

type MultiLoginPolicy struct {
	// AllowMultiLogin 是否允许同一用户多端在线（false：新连接踢掉全部旧连接）
	AllowMultiLogin bool
	// MaxConnectionsPerUser 单用户连接数上限（0 表示不限制；达上限踢最旧连接）
	MaxConnectionsPerUser int
}

MultiLoginPolicy 多端登录策略快照（编排层构造期从全局配置提取）

type RecordHost 

type RecordHost interface {
	// GetLogger 日志器
	GetLogger() spi.Logger
	// GetConnectionRecordRepo 连接记录仓储（未注入返回 nil，调用方自行降级）
	GetConnectionRecordRepo() spi.ConnectionStore
	// GetConnectionQualityRepository 连接质量仓储（未注入返回 nil，调用方自行降级）：
	// 注册落库时与 connect 行 1:1 同生 quality 初始行，batcher 的心跳/统计/错误
	// 批量 UPDATE 才有落点（无初始行则全部空转影响 0 行）
	GetConnectionQualityRepository() spi.ConnectionQualityStore
	// GetDisconnectionBatcher 断连终态攒批器（未注入返回 nil，调用方自行降级）
	GetDisconnectionBatcher() DisconnectionSubmitter
}

RecordHost 连接记录子域所需的 Hub 能力面

仓储经端口动态读取而非构造期快照：SetConnectionRecordRepository 支持 运行期装配（NewHub 后经 adapter 注入），每次读写实时反映注入状态

type RecordManager 

type RecordManager struct {
	// contains filtered or unexported fields
}

RecordManager 连接记录管理器（连接域）

覆盖连接记录的构造（Client → ConnectionRecord 快照）与终态写路径 （注册异步保存 / 注销异步标记断开 / 停机批量终态）

func NewRecordManager 

func NewRecordManager(host RecordHost) *RecordManager

NewRecordManager 构造连接记录管理器

func (*RecordManager) Create 

func (m *RecordManager) Create(client *models.Client) *models.ConnectionRecord

Create 构造连接记录（内存对象，供异步保存 + 连接回调使用）

func (*RecordManager) MarkDisconnected 

func (m *RecordManager) MarkDisconnected(client *models.Client)

MarkDisconnected 标记连接为已断开（攒批路径，batcher 未注入时 no-op） 构造 DisconnectionEntry 快照提交到 DisconnectionBatcher，由后台攒批合并为 CASE WHEN 单 SQL 落库；快照在断连瞬间冻结 DisconnectedAt/ConnectedAt， flush 延迟不虚增 duration；队列满丢弃与原记录池可丢弃语义一致

func (*RecordManager) MarkDisconnectedBatch 

func (m *RecordManager) MarkDisconnectedBatch(clients []*models.Client) []*models.DisconnectionEntry

MarkDisconnectedBatch 停机批量标记连接断开（仓储未注入时返回 nil） 与单连接路径的差异：reason 为 ServerShutdown + 关闭码 1001（客户端据此识别服务端主动离开并重连）； 不走攒批队列而是分块同步直调 BatchMarkDisconnected—— 停机路径需在本方法返回后立即被批量终评读 duration 算分，保序优先 返回构造的 entries 供终评直接复用 duration（与 SQL 写入值同源同公式），免去终评再把刚落库的 duration 读回来的往返

func (*RecordManager) Save 

func (m *RecordManager) Save(ctx context.Context, record *models.ConnectionRecord)

Save 保存或更新连接记录到数据库（仓储未注入时 no-op） ctx 应为 client.Context（带 client 维度的 trace_id），实现异步保存的全链路追踪 connect 身份行与 quality 初始行同批落库：batcher 的心跳/统计/错误批量 UPDATE 以 quality 行存在为前提，无初始行则全部空转影响 0 行（曾致质量表长期零数据）

type RegistryCapacity 

type RegistryCapacity struct {
	// TotalClients 总客户端连接数预估（主存储 userShards）
	TotalClients int
	// SSEClients SSE 客户端连接数预估（sseShards）
	SSEClients int
	// ObserverClients 观察者连接数预估（observerShards）
	ObserverClients int
	// AgentClients 客服/Bot 连接数预估（agentShards）
	AgentClients int
}

RegistryCapacity 容量提示（用于预分配各 ShardedMap 的每 shard 内部 map 容量）

设计思路：

  • 总容量除以 shardCount 得到每 shard 提示，减少 map 扩容次数
  • 各分类索引（SSE/Observer/Agent）按预估占比独立分配，避免过度预分配
  • 0 表示不预分配，由 map 自适应扩容

type ShardedRegistry 

type ShardedRegistry struct {
	// contains filtered or unexported fields
}

ShardedRegistry 分片注册表 基于 syncx.ShardedMap 实现，替代 Hub 的单一 mutex + map 结构 同时承担原本由 Hub 维护的 sseClients/observerClients/agentClients 分类索引

func NewShardedRegistry 

func NewShardedRegistry(agentEnabled, observerEnabled bool, capacity RegistryCapacity) *ShardedRegistry

NewShardedRegistry 创建分片注册表 agentEnabled / observerEnabled 控制是否启用对应分类分片； 未启用的分类 shards 保持 nil，相关读写直接跳过，避免无谓内存与锁开销

capacity 提供总容量预估，用于预分配各 ShardedMap 的每 shard 内部 map 容量； 传入零值 RegistryCapacity 时退化为不预分配（与旧版兼容）

func (*ShardedRegistry) AddClient 

func (r *ShardedRegistry) AddClient(client *models.Client)

AddClient 添加客户端到注册表 主存储 + 分类索引在同一 userID 的 shard 锁内原子完成（ShardedMap 保证同一 key 落同一 shard）

覆盖语义（断线重连场景）：当相同 clientID 已存在时，仅更新主存储与分类索引的指针， 不重复累加 clientCount/sseCount，否则计数器会随每次重连持续膨胀、与实际连接数脱节 observerUserCount/agentUserCount 在各自 add 方法内已按"用户是否新增"判定，无需此处处理

func (*ShardedRegistry) AgentEnabled 

func (r *ShardedRegistry) AgentEnabled() bool

AgentEnabled 返回客服模块是否启用（shards 非 nil）

func (*ShardedRegistry) Clear 

func (r *ShardedRegistry) Clear()

Clear 清空注册表

func (*ShardedRegistry) FilterClients 

func (r *ShardedRegistry) FilterClients(predicate func(*models.Client) bool) []*models.Client

FilterClients 按条件过滤客户端（零拷贝：ForEachClient 遍历 + 条件收集，避免 GetClientsCopy 全量拷贝）

func (*ShardedRegistry) ForEachClient 

func (r *ShardedRegistry) ForEachClient(fn func(clientID string, client *models.Client) bool)

ForEachClient 遍历所有客户端 回调返回 false 时停止遍历 注意：回调在持有读锁时执行，不应执行耗时操作

func (*ShardedRegistry) ForEachClientFiltered 

func (r *ShardedRegistry) ForEachClientFiltered(msgAppID, msgNamespace string, msgGroupIDs []string, fn func(clientID string, client *models.Client) bool)

ForEachClientFiltered 遍历注册表所有 client，仅对匹配路由信封(含 appId)的 client 调用 fn 用于全局广播等场景：app1/ns1 的广播只投递给 app1/ns1 的所有在线设备

func (*ShardedRegistry) ForEachClientFilteredParallel 

func (r *ShardedRegistry) ForEachClientFilteredParallel(workerNum int, msgAppID, msgNamespace string, msgGroupIDs []string, fn func(clientID string, client *models.Client))

ForEachClientFilteredParallel 并行遍历所有客户端，仅对匹配路由信封(含 appId)的 client 调用 fn 用于全局广播等场景：app1/ns1 的广播只投递给 app1/ns1 的所有在线设备 基于 userShards.RangeParallel 直接在 shard 级别并行遍历，零额外分配 workerNum <= 0 时使用 GOMAXPROCS；用户数低于 parallelThreshold 时退化为串行

func (*ShardedRegistry) ForEachClientParallel 

func (r *ShardedRegistry) ForEachClientParallel(workerNum int, fn func(clientID string, client *models.Client))

ForEachClientParallel 并行遍历所有客户端（百万级连接优化） 基于 userShards.RangeParallel 直接在 shard 级别并行遍历，零额外分配 fn 在 shard 读锁内执行（并发安全），回调应为非阻塞操作 workerNum <= 0 时使用 GOMAXPROCS；用户数低于 parallelThreshold 时退化为串行

func (*ShardedRegistry) ForEachObserver 

func (r *ShardedRegistry) ForEachObserver(fn func(userID string, clientID string, client *models.Client) bool)

ForEachObserver 遍历所有观察者客户端

func (*ShardedRegistry) ForEachSSEClient 

func (r *ShardedRegistry) ForEachSSEClient(fn func(userID string, clientID string, client *models.Client) bool)

ForEachSSEClient 遍历所有 SSE 客户端（用于广播） 回调返回 false 时停止遍历

func (*ShardedRegistry) ForEachSSEClientParallel 

func (r *ShardedRegistry) ForEachSSEClientParallel(workerNum int, fn func(userID, clientID string, client *models.Client))

ForEachSSEClientParallel 并行遍历所有 SSE 客户端（百万级连接优化） 基于 sseShards.RangeParallel 直接在 shard 级别并行遍历，零额外分配 workerNum <= 0 时使用 GOMAXPROCS；用户数低于 parallelThreshold 时退化为串行

func (*ShardedRegistry) ForEachSSEUserClient 

func (r *ShardedRegistry) ForEachSSEUserClient(userID string, fn func(clientID string, client *models.Client) bool)

ForEachSSEUserClient 遍历指定用户的所有 SSE 客户端（持读锁，零拷贝） 回调返回 false 时停止遍历 替代 GetSSEUserClients 锁外遍历内部 map 的数据竞争

func (*ShardedRegistry) ForEachUser 

func (r *ShardedRegistry) ForEachUser(fn func(userID string, clients map[string]*models.Client) bool)

ForEachUser 遍历所有在线用户 回调返回 false 时停止遍历 注意：回调在持有读锁时执行，不应执行耗时操作

func (*ShardedRegistry) ForEachUserClient 

func (r *ShardedRegistry) ForEachUserClient(userID string, fn func(clientID string, client *models.Client) bool)

ForEachUserClient 遍历指定用户的所有客户端（在读锁内执行，并发安全） 回调返回 false 时停止遍历回调不应执行阻塞操作（TrySend 等非阻塞操作可安全调用）

func (*ShardedRegistry) ForEachUserClientFiltered 

func (r *ShardedRegistry) ForEachUserClientFiltered(userID, msgAppID, msgNamespace string, msgGroupIDs []string, fn func(clientID string, client *models.Client) bool)

ForEachUserClientFiltered 遍历指定 userID 的所有在线设备，仅对匹配路由信封(含 appId)的 client 调用 fn fn 返回 false 时中止遍历（与 ForEachUserClient 语义一致） 用于 P2P/群组的定向投递：即使同名 userID 跨 app/namespace 同时在线，也只投递给路由匹配的设备

appId/namespace 匹配规则（复用 ClientMatchesEnvelope，保持单一真相源）：

  • appId 严格相等（入口层已归一化为 DefaultAppID，无空值兼容）
  • namespace 非空时严格相等；msgNamespace=="" 表示全局广播，跳过 ns 过滤匹配所有

设计说明：只做 appId+namespace 隔离，**不对 msgGroupIDs vs client.GroupID 做系统组匹配**

  • 场景1（P2P 点对点）：调用方已明确指定 userID 为接收者，msg.GroupIDs=nil → 无 group 维度
  • 场景2（群组消息）：调用方已通过 group_repo.GetMembers() 获取业务群组成员 userIDs， msg.GroupIDs 存放的是"业务群组ID"（如 "g-broadcast"），而 client.GroupID 是"连接级系统组" （如 __default_gp__），两者完全两个维度，强行匹配导致群成员 device 全部被过滤（delivered=0）

func (*ShardedRegistry) GetActiveClientCount 

func (r *ShardedRegistry) GetActiveClientCount() int64

GetActiveClientCount 获取活跃（非 SSE）客户端数 = 总连接数 - SSE 连接数

func (*ShardedRegistry) GetAgentUserClients 

func (r *ShardedRegistry) GetAgentUserClients(userID string) (map[string]*models.Client, bool)

GetAgentUserClients 获取指定客服的所有客户端

func (*ShardedRegistry) GetAgentUserCount 

func (r *ShardedRegistry) GetAgentUserCount() int

GetAgentUserCount 获取客服用户数 - O(1)

func (*ShardedRegistry) GetAllClients 

func (r *ShardedRegistry) GetAllClients() []*models.Client

GetAllClients 获取所有客户端列表 通过 ShardedMap.Range 遍历所有 shard（分片读锁，粒度细）

func (*ShardedRegistry) GetClient 

func (r *ShardedRegistry) GetClient(clientID string) (*models.Client, bool)

GetClient 根据 clientID 获取客户端 先查 clientIDToUserID 反向索引定位 shard，再读锁查找

func (*ShardedRegistry) GetClientCount 

func (r *ShardedRegistry) GetClientCount() int64

GetClientCount 获取客户端总数（原子读取，零锁开销）

func (*ShardedRegistry) GetClientIPs 

func (r *ShardedRegistry) GetClientIPs(ctx context.Context, userID string) []string

GetClientIPs 获取用户所有客户端的IP地址列表（按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入，调用方不需传 appID/namespace 参数 使用 ForEachUserClientFiltered 零拷贝遍历（复用 ClientMatchesEnvelope 单一真相源）

func (*ShardedRegistry) GetClientMetadata 

func (r *ShardedRegistry) GetClientMetadata(clientID string, key string) (interface{}, bool)

GetClientMetadata 获取客户端元数据（线程安全）

func (*ShardedRegistry) GetClientStats 

func (r *ShardedRegistry) GetClientStats(clientID string) map[string]interface{}

GetClientStats 获取客户端统计信息

func (*ShardedRegistry) GetClientsByDepartmentGrouped 

func (r *ShardedRegistry) GetClientsByDepartmentGrouped() map[models.Department][]*models.Client

GetClientsByDepartmentGrouped 按部门分组获取客户端（零拷贝遍历）

func (*ShardedRegistry) GetClientsByStatusGrouped 

func (r *ShardedRegistry) GetClientsByStatusGrouped() map[models.UserStatus][]*models.Client

GetClientsByStatusGrouped 按状态分组获取客户端（零拷贝遍历，原子读 Status）

func (*ShardedRegistry) GetClientsByUserID 

func (r *ShardedRegistry) GetClientsByUserID(ctx context.Context, userID string) []*models.Client

GetClientsByUserID 根据用户ID获取所有客户端（按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入，调用方不需传 appID/namespace 参数 使用 ForEachUserClientFiltered 持读锁遍历（复用 ClientMatchesEnvelope 单一真相源）

func (*ShardedRegistry) GetClientsByUserTypeGrouped 

func (r *ShardedRegistry) GetClientsByUserTypeGrouped() map[models.UserType][]*models.Client

GetClientsByUserTypeGrouped 按用户类型分组获取客户端（零拷贝遍历）

func (*ShardedRegistry) GetClientsCopyForUser 

func (r *ShardedRegistry) GetClientsCopyForUser(ctx context.Context, userID, clientID string) []*models.Client

GetClientsCopyForUser 获取用户的客户端列表副本（线程安全，按 ctx 路由信封 appID+namespace 隔离） 如果指定了 clientID，只返回该客户端（O(1) 查找，仍校验 client 的 appID+namespace 信封匹配） 否则返回用户匹配信封的所有客户端 使用 ForEachUserClientFiltered 持读锁遍历，替代 GetUserClients 锁外遍历的数据竞争

func (*ShardedRegistry) GetClientsWithStatus 

func (r *ShardedRegistry) GetClientsWithStatus(status models.UserStatus) []*models.Client

GetClientsWithStatus 获取指定状态的所有客户端（委托 FilterClients 零拷贝，原子读 Status）

func (*ShardedRegistry) GetLocalGroupMembers 

func (r *ShardedRegistry) GetLocalGroupMembers(appID string, groupIDs []string) []string

GetLocalGroupMembers 返回指定 appID 下多群组的本地在线成员 userID 列表（跨群组去重） 替代 Redis GetMultiGroupMembers 的全量读：各节点仅返回本节点在线成员，扇出范围天然收敛为本地连接 返回列表不含 namespace 维度（群组投递跨 ns 通配），下游 BroadcastToUserIDs 按信封 appID 过滤

func (*ShardedRegistry) GetLocalNamespaceUsers 

func (r *ShardedRegistry) GetLocalNamespaceUsers(appID, namespace string) []string

GetLocalNamespaceUsers 返回指定 appID+namespace 下的本地在线用户 userID 列表（设备去重） 命名空间广播（本地投递与跨节点消费侧）经此定点反查后 BroadcastToUserIDs 扇出， 替代 BroadcastToFiltered 对全部连接的 O(N) 全量扫描； namespace 为空时返回 nil：空 ns = 全局广播语义，由全量扫描路径处理（目标集=全部连接，全量即最优）， 不允许从本索引取数（索引只按非空 ns 维度建键）

func (*ShardedRegistry) GetLocalUsersByClientType 

func (r *ShardedRegistry) GetLocalUsersByClientType(appID string, clientType models.ClientType) ([]string, bool)

GetLocalUsersByClientType 返回指定客户端类型的本地在线用户（设备去重）

func (*ShardedRegistry) GetLocalUsersByDepartment 

func (r *ShardedRegistry) GetLocalUsersByDepartment(appID string, dept models.Department) ([]string, bool)

GetLocalUsersByDepartment 返回指定部门的本地在线用户（设备去重）

func (*ShardedRegistry) GetLocalUsersByRole 

func (r *ShardedRegistry) GetLocalUsersByRole(appID string, role models.UserRole) ([]string, bool)

GetLocalUsersByRole 返回指定角色的本地在线用户（设备去重）

func (*ShardedRegistry) GetLocalUsersByUserType 

func (r *ShardedRegistry) GetLocalUsersByUserType(appID string, userType models.UserType) ([]string, bool)

GetLocalUsersByUserType 返回指定用户类型的本地在线用户（设备去重） indexed=false 表示未入索引（空取值），调用方须回退全量扫描

func (*ShardedRegistry) GetMostRecentClient 

func (r *ShardedRegistry) GetMostRecentClient(ctx context.Context, userID string) *models.Client

GetMostRecentClient 获取用户对应的客户端（返回最近活跃的客户端，按 ctx 路由信封 appID+namespace 隔离） 使用 ForEachUserClientFiltered 零拷贝遍历（持读锁），替代 GetUserClients 锁外遍历内部 map 的数据竞争

func (*ShardedRegistry) GetObserverDeviceCount 

func (r *ShardedRegistry) GetObserverDeviceCount() int

GetObserverDeviceCount 获取观察者设备总数

func (*ShardedRegistry) GetObserverUserClients 

func (r *ShardedRegistry) GetObserverUserClients(userID string) (map[string]*models.Client, bool)

GetObserverUserClients 获取指定观察者的所有客户端

func (*ShardedRegistry) GetObserverUserCount 

func (r *ShardedRegistry) GetObserverUserCount() int

GetObserverUserCount 获取观察者用户数（设备去重）- O(1)

func (*ShardedRegistry) GetObserversForMessage 

func (r *ShardedRegistry) GetObserversForMessage(namespace string, groupIDs ...string) []*models.Client

func (*ShardedRegistry) GetOnlineUserCountByType 

func (r *ShardedRegistry) GetOnlineUserCountByType(userType models.UserType) (int64, error)

GetOnlineUserCountByType 根据用户类型获取在线用户数量 使用 ForEachClient 零拷贝遍历 + 用户去重，替代 GetClientsByUserType 切片拷贝

func (*ShardedRegistry) GetOnlineUserIDs 

func (r *ShardedRegistry) GetOnlineUserIDs() []string

GetOnlineUserIDs 获取所有在线用户ID列表 使用 ShardedMap.Keys 直接获取所有 userID

func (*ShardedRegistry) GetOnlineUsersByType 

func (r *ShardedRegistry) GetOnlineUsersByType(userType models.UserType) ([]string, error)

GetOnlineUsersByType 按用户类型获取在线用户列表 使用 ForEachClient 零拷贝遍历 + 用户去重，替代 FilterClients 切片拷贝

func (*ShardedRegistry) GetSSEClientCount 

func (r *ShardedRegistry) GetSSEClientCount() int64

GetSSEClientCount 获取 SSE 客户端总数（原子读取，零锁开销）

func (*ShardedRegistry) GetSSEUserClients 

func (r *ShardedRegistry) GetSSEUserClients(userID string) (map[string]*models.Client, bool)

GetSSEUserClients 获取指定用户的所有 SSE 客户端（用于按 userID 推送） 返回内部 map 引用，调用方应在读锁释放后立即使用或拷贝

func (*ShardedRegistry) GetSSEUserCount 

func (r *ShardedRegistry) GetSSEUserCount() int

GetSSEUserCount 获取 SSE 在线用户数

func (*ShardedRegistry) GetSSEUserIDs 

func (r *ShardedRegistry) GetSSEUserIDs() []string

GetSSEUserIDs 获取所有 SSE 在线用户ID

func (*ShardedRegistry) GetUserClientCount 

func (r *ShardedRegistry) GetUserClientCount(userID string) int

GetUserClientCount 获取用户的客户端数量

func (*ShardedRegistry) GetUserClients 

func (r *ShardedRegistry) GetUserClients(userID string) (map[string]*models.Client, bool)

GetUserClients 获取用户的所有客户端 返回的是内部 map 引用（读锁期间获取的快照），调用方不应长时间持有

func (*ShardedRegistry) GetUserClientsCopy 

func (r *ShardedRegistry) GetUserClientsCopy() []*models.Client

GetUserClientsCopy 获取每个用户最活跃的客户端副本列表 遍历所有 shard，每个用户取最近活跃的客户端（复用 findMostRecentClient）

func (*ShardedRegistry) GetUserCount 

func (r *ShardedRegistry) GetUserCount() int64

GetUserCount 获取在线用户总数（原子读取，零锁开销）

func (*ShardedRegistry) GetUserStatus 

func (r *ShardedRegistry) GetUserStatus(ctx context.Context, userID string) models.UserStatus

GetUserStatus 获取用户状态（按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入，调用方不需传 appID/namespace 参数 使用 ForEachUserClientFiltered 零拷贝遍历（复用 ClientMatchesEnvelope 单一真相源）

func (*ShardedRegistry) HasAgent 

func (r *ShardedRegistry) HasAgent(userID string) bool

HasAgent 检查用户是否为客服 - O(1)

func (*ShardedRegistry) HasObserver 

func (r *ShardedRegistry) HasObserver(userID string) bool

HasObserver 检查用户是否为观察者 - O(1)

func (*ShardedRegistry) HasSSEUser 

func (r *ShardedRegistry) HasSSEUser(userID string) bool

HasSSEUser 检查指定用户是否有 SSE 连接

func (*ShardedRegistry) HasUser 

func (r *ShardedRegistry) HasUser(userID, appID, namespace string) bool

HasUser 检查指定路由信封(appID+namespace)下某用户是否在线（统一入口，强制 appID+namespace 隔离）

设计原则：appID+namespace 是隔离维度（查询/投递一律走信封），group 是广播维度（与本方法正交） 路由信封来源统一为 ctx（Hub 层 routing.AppIDFromContext/NamespaceFromContext 提取后传入）

复用 ClientMatchesEnvelope 单一真相源：appID 严格匹配 + namespace 空值兼容全局广播 appID 为空时退化为不过滤（兼容无路由 ctx 的边界场景），正常路径入口已归一化为 DefaultAppID 性能：单次读锁内完成存在性检查 + 信封过滤（map miss 即 O(1) 判负，在线场景 2 锁 → 1 锁）

func (*ShardedRegistry) HasUserClient 

func (r *ShardedRegistry) HasUserClient(ctx context.Context, userID string) bool

HasUserClient 检查是否存在指定用户ID的客户端（按 ctx 路由信封的 appID+namespace 隔离） 路由信封通过 routing 从 ctx 注入，调用方不需传 appID/namespace 参数

func (*ShardedRegistry) HasUserExactEnvelope 

func (r *ShardedRegistry) HasUserExactEnvelope(userID, appID, namespace string) bool

HasUserExactEnvelope 检查指定信封(appID+namespace 原值)下某用户是否在线（分区精确匹配）

与 HasUser 的区别：namespace 空值不做全局放宽——HasUser 的空值语义服务于广播投递范围 （namespace=="" 匹配所有命名空间），本方法服务于分区生命周期判定（系统组成员 join/leave 必须读写同一分区 key，"" 分区不因用户在其他命名空间在线而误判为仍需保留）

func (*ShardedRegistry) ObserverEnabled 

func (r *ShardedRegistry) ObserverEnabled() bool

ObserverEnabled 返回观察者模块是否启用（shards 非 nil）

func (*ShardedRegistry) RemoveClient 

func (r *ShardedRegistry) RemoveClient(clientID, userID string) *models.Client

RemoveClient 从注册表移除客户端 返回被移除的客户端（nil 表示不存在）

func (*ShardedRegistry) UpdateClientMetadata 

func (r *ShardedRegistry) UpdateClientMetadata(clientID string, key string, value interface{}) error

UpdateClientMetadata 更新客户端元数据（线程安全）

type SlowConsumerScanner 

type SlowConsumerScanner struct {
	// contains filtered or unexported fields
}

SlowConsumerScanner 慢消费者扫描器（周期采样 + 三级治理）

func NewSlowConsumerScanner 

func NewSlowConsumerScanner(registry *ShardedRegistry, hook EvictHook, interval time.Duration, logger spi.Logger) *SlowConsumerScanner

NewSlowConsumerScanner 创建慢消费者扫描器 interval ≤ 0 时默认 1s（与原逻辑"准入未配置时 1s"对齐）

func (*SlowConsumerScanner) ScanOnce 

func (s *SlowConsumerScanner) ScanOnce()

ScanOnce 单轮扫描（分片并行遍历 + 无锁采样 + 三级治理） 导出供测试与运维接口手动触发；并发手动触发与周期扫描间整轮互斥

func (*SlowConsumerScanner) Start 

func (s *SlowConsumerScanner) Start(ctx context.Context)

Start 启动周期扫描（阻塞直至 ctx 取消；Hub Run 时以 go 调用）

周期 = 准入评估周期 × 2（与水位评估同源节拍，避免扫描与评估完全同步造成的 周期性毛刺）；扫描本身 O(活跃连接) 原子读遍历

func (*SlowConsumerScanner) Stats 

func (s *SlowConsumerScanner) Stats() int

Stats 治理状态统计（观测用：当前处于积压追踪中的连接数）

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.