hub

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: 40 Imported by: 0

README

hub（编排层）

组装 7 个域、持有全部消费者端口的实现、对外暴露唯一入口——只编排，不实现

这个域干什么

hub 是 go-wsc 的对外唯一入口它把 connection / transport / messaging / group / stats / cluster / overload 七个域装配成一个可运行的 Hub，并实现各域声明的 「消费者端口」（每个子包的 ports.go）

管：域的装配与生命周期（Run / SafeShutdown / WaitForStart）、端口实现集中处 （ports.go）、对外 API 的签名聚合、context key 定义

不管：任何业务逻辑注册表在 connection，投递在 messaging，削峰在 overload—— hub 里出现 for 循环处理消息就是设计倒退

核心概念

概念 说明
Hub 编排器本体，~30 字段（各域 Manager 的引用），无业务方法
NewHub(cfg) 构造入口，按配置装配各域；零注入时各域为纯内存实现
消费者端口 子包在 ports.go 声明最小接口，Hub 集中实现（见 hub/ports.go）
With* 选项 WithOnlineStore / WithMessageSink 等注入后端适配器
Run / SafeShutdown 启动各域后台任务 / 排空并停止（联动 K8s SIGTERM drain）

怎么用

package main

import (
    "net/http"

    "github.com/kamalyes/go-wsc/hub"
)

func main() {
    h := hub.NewHub(cfg)          // 纯内存单节点：零注入即可跑
    // h := hub.NewHub(cfg, hub.WithOnlineStore(redisAdapter))  // 接后端是显式选择

    go h.Run()
    defer h.SafeShutdown()

    // 框架无关：标准 http 签名，随便嵌哪个路由
    http.HandleFunc("/ws", h.HandleWebSocketUpgrade)
    http.HandleFunc("/sse", h.HandleSSEUpgrade)
    http.ListenAndServe(":8080", nil)
}

端口与依赖

  • hub 实现所有子包的消费者端口（batcher.StorageBatchWriter、group.Host、stats.Host 等）
  • 只依赖 spi 契约与各域 Manager，零 gorm / redis / grpc / nats / clickhouse import
  • 铁律：grep -rn "middleware.|repository." hub/ 必须为空（详见 REFACTORING_DESIGN §一.1）

性能与陷阱

  • hub 本身无热路径——百万连接的吞吐在各域内部，hub 只做指针转发
  • 陷阱：往 Hub 上继续堆方法就是回到 74 字段的上帝对象老路；新能力先进对应域，hub 只加端口

后端绑定

纯编排，无后端所有 With* 注入的都是适配器仓库（go-wsc-redis-adapter 等）的实例

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AfterHeartbeatCallback 

type AfterHeartbeatCallback func(client *models.Client)

AfterHeartbeatCallback 心跳处理后回调

type BeforeHeartbeatCallback 

type BeforeHeartbeatCallback func(client *models.Client) bool

BeforeHeartbeatCallback 心跳处理前回调，返回 false 则跳过后续心跳处理

type ClientConnectCallback 

type ClientConnectCallback func(ctx context.Context, client *models.Client, record *models.ConnectionRecord) error

ClientConnectCallback 客户端连接回调

type ClientDisconnectCallback 

type ClientDisconnectCallback func(ctx context.Context, client *models.Client, reason models.DisconnectReason) error

ClientDisconnectCallback 客户端断开回调

type GRPCServer 

type GRPCServer struct {
	wscpb.UnimplementedNodeServiceServer
	// contains filtered or unexported fields
}

GRPCServer gRPC 服务端，实现 wscpb.NodeServiceServer 接口 将远端节点的 gRPC 请求转交给本地 Hub 处理，实现跨节点精确路由

func NewGRPCServer 

func NewGRPCServer(hub *Hub) *GRPCServer

NewGRPCServer 创建新的 gRPC 服务端

func (*GRPCServer) BatchDispatch 

func (s *GRPCServer) BatchDispatch(ctx context.Context, req *wscpb.BatchDispatchRequest) (*wscpb.BatchDispatchResponse, error)

BatchDispatch 跨节点微批合帧投递（多条指令合并为单次 RPC）

逐条委托 dispatchBatchItem 处理，结果与请求 items 一一对应； 单条失败不中断整批（结果经 DispatchResult.Error 回传，由发送端逐条降级 PubSub）

func (*GRPCServer) BroadcastGroup 

func (s *GRPCServer) BroadcastGroup(ctx context.Context, req *wscpb.BroadcastGroupRequest) (*wscpb.BroadcastGroupResponse, error)

BroadcastGroup 向本节点的群组成员广播消息 生产者-消费者模式：消费端经本地反向索引反查本节点成员后扇出，替代 Redis GetMultiGroupMembers 全量读

func (*GRPCServer) CheckUsersOnline 

func (s *GRPCServer) CheckUsersOnline(ctx context.Context, req *wscpb.CheckUsersOnlineRequest) (*wscpb.CheckUsersOnlineResponse, error)

CheckUsersOnline 批量检查用户是否在本节点在线（路由探测） 使用 HasUserClient O(1) 检查，替代 GetClientsByUserID 切片分配 按 ctx 路由信封 appID+namespace 隔离（路由从 gRPC metadata 恢复）

func (*GRPCServer) KickUser 

func (s *GRPCServer) KickUser(ctx context.Context, req *wscpb.KickUserRequest) (*wscpb.KickUserResponse, error)

KickUser 踢出本节点上的用户 路由信封从 gRPC incoming metadata 恢复（appID+namespace），按信封隔离踢出同名用户连接

func (*GRPCServer) NotifyObservers 

func (s *GRPCServer) NotifyObservers(ctx context.Context, req *wscpb.NotifyObserversRequest) (*wscpb.NotifyObserversResponse, error)

NotifyObservers 通知本节点的观察者 namespace/groupID 从 gRPC incoming metadata 恢复到 ctx 后提取

func (*GRPCServer) Ping 

func (s *GRPCServer) Ping(ctx context.Context, req *wscpb.PingRequest) (*wscpb.PingResponse, error)

Ping 节点健康检查

func (*GRPCServer) SendToUser 

func (s *GRPCServer) SendToUser(ctx context.Context, req *wscpb.SendToUserRequest) (*wscpb.SendToUserResponse, error)

SendToUser 向本节点的指定用户发送消息（点对点投递） 使用 ForEachUserClientFiltered 零拷贝遍历 + 预序列化，替代 GetClientsByUserID 切片拷贝 + 逐客户端序列化

func (*GRPCServer) Start 

func (s *GRPCServer) Start(ctx context.Context, addr string) error

Start 启动 gRPC 服务端，监听指定地址并异步提供服务 addr 支持 IPv4（host:port）和 IPv6（[host]:port 或裸 IPv6 地址）

func (*GRPCServer) Stop 

func (s *GRPCServer) Stop()

Stop 优雅停止 gRPC 服务端

type HeartbeatReportCallback 

type HeartbeatReportCallback func(client *models.Client)

HeartbeatReportCallback 心跳上报回调

type HeartbeatTimeoutCallback 

type HeartbeatTimeoutCallback func(clientID string, userID string, lastHeartbeat time.Time)

HeartbeatTimeoutCallback 心跳超时回调

type Hub 

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

Hub WebSocket/SSE 连接管理中心（编排层）

持有域管理器（messaging/stats/group）与运行时组件（overload/cluster/connection）， 实现各域 Host 端口做委托。域逻辑不在本结构内，仅做组装与转发。

func NewHub 

func NewHub(config *wscconfig.WSC) *Hub

NewHub 创建新的 Hub（编排层组装入口）

域管理器与批处理器均以 hub 作为 Host 端口构造：Go 允许在构造函数返回前 使用指针的方法集，因为方法集在编译期已确定。

func (*Hub) AcquireDistributedLock 

func (h *Hub) AcquireDistributedLock(ctx context.Context, key string, ttl time.Duration) (bool, error)

AcquireDistributedLock 获取分布式锁

func (*Hub) AdmissionOnDelivered 

func (h *Hub) AdmissionOnDelivered()

AdmissionOnDelivered 投递完成埋点（闸门未启用时 no-op）

func (*Hub) AdmitMessage 

func (h *Hub) AdmitMessage(msg *models.HubMessage, isBroadcast bool) overload.AdmitVerdict

AdmitMessage 消息级准入裁决（闸门未启用时恒放行） 热路径契约：零分配（枚举返回）；isBroadcast 标识广播路径（L2 阶梯差异化）

func (*Hub) BatchGetUserNodes 

func (h *Hub) BatchGetUserNodes(ctx context.Context, userIDs []string) map[string][]string

BatchGetUserNodes 批量查询多个用户所在节点（Pipeline 单次往返，信封过滤与路由上下文一致） 消息域唯一节点查询端口（单用户传单元素切片复用）；查询失败返回 nil，调用方回退逐用户查询

func (*Hub) BroadcastToSSEClients 

func (h *Hub) BroadcastToSSEClients(msg *models.HubMessage)

BroadcastToSSEClients 广播给全部 SSE 客户端（appId/namespace 信封隔离） 通过 ForEachSSEClientParallel 并行分片读锁遍历（百万级优化）

func (*Hub) CheckAndRouteToNode 

func (h *Hub) CheckAndRouteToNode(ctx context.Context, userID string, msg *models.HubMessage, presetNodes []string) (bool, []string, error)

CheckAndRouteToNode 检查用户在线节点并按需跨节点投递 presetNodes 非空时跳过节点查询（调用方已预取） 返回：是否命中跨节点路由、目标节点列表、错误

func (*Hub) CheckUserOnline 

func (h *Hub) CheckUserOnline(ctx context.Context, userID string) bool

CheckUserOnline 检查用户是否在线（跨节点汇总判定；对外诊断 API， 消息发送热路径已合并为 GetUserNodes 单次往返，不再走此方法）

func (*Hub) Context 

func (h *Hub) Context() context.Context

Context 返回 Hub 生命周期上下文

func (*Hub) DeferBroadcast 

func (h *Hub) DeferBroadcast(ctx context.Context, msg *models.HubMessage, retry func(context.Context, *models.HubMessage))

DeferBroadcast 广播延迟重投（延迟队列，拒绝不等于丢弃）

func (*Hub) DeleteRerouteGuard 

func (h *Hub) DeleteRerouteGuard(messageID string)

DeleteRerouteGuard 清理重路由守卫条目（消息到达终态时调用）

func (*Hub) Deliver 

func (h *Hub) Deliver(ctx context.Context, msg *models.HubMessage, excludeSender bool) *models.DeliverResult

Deliver 统一投递入口（P2P / 群组 / 广播决策树，委托消息域） 路由维度（appID/namespace/groupIDs）经 routing.NewRoute() 链式构建器注入 ctx：

  • P2P：msg.Receiver 非空触发（在线投递 + 离线存储 + 重试）
  • 群组可靠投递：msg.RequireAck=true 触发（per-member 重试 + 离线存储）
  • 命名空间/全局广播：ctx 信封 namespace 决策（空=全局）

excludeSender：群组/广播场景排除发送者自身连接

func (*Hub) EnqueueHeartbeatRenew 

func (h *Hub) EnqueueHeartbeatRenew(client *models.Client)

EnqueueHeartbeatRenew 心跳 Redis 在线索引续期入队（connection.HeartbeatHost 端口） 单 goroutine worker 消费 heartbeatRedisCh，满则丢弃（心跳下次还会来）

func (*Hub) FlushStats 

func (h *Hub) FlushStats()

FlushStats 公共方法：立即将内存中累积的消息/广播统计计数器刷写到 Redis 正常运行时由 30 秒定时器自动刷写，测试或需要即时统计的场景可手动调用

func (*Hub) GetAdmissionLevel 

func (h *Hub) GetAdmissionLevel() overload.OverloadLevel

GetAdmissionLevel 当前过载水位（闸门未启用时返回 LevelNormal）

func (*Hub) GetAfterHeartbeatCallback 

func (h *Hub) GetAfterHeartbeatCallback() func(client *models.Client)

GetAfterHeartbeatCallback 心跳后置回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetAllClusterNodeIDs 

func (h *Hub) GetAllClusterNodeIDs() []string

GetAllClusterNodeIDs 获取集群全部节点 ID

func (*Hub) GetBeforeHeartbeatCallback 

func (h *Hub) GetBeforeHeartbeatCallback() func(client *models.Client) bool

GetBeforeHeartbeatCallback 心跳前置回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetBroadcastShaper 

func (h *Hub) GetBroadcastShaper() *overload.Shaper

GetBroadcastShaper 广播出向整形器（未启用时返回 nil）

func (*Hub) GetConfig 

func (h *Hub) GetConfig() *wscconfig.WSC

GetConfig 全局配置

func (*Hub) GetConnectionQualityRepository 

func (h *Hub) GetConnectionQualityRepository() spi.ConnectionQualityStore

GetConnectionQualityRepository 连接质量仓储

func (*Hub) GetConnectionRecordRepo 

func (h *Hub) GetConnectionRecordRepo() spi.ConnectionStore

GetConnectionRecordRepo 连接记录仓储

func (*Hub) GetDisconnectionBatcher 

func (h *Hub) GetDisconnectionBatcher() connection.DisconnectionSubmitter

GetDisconnectionBatcher 断连终态攒批器（连接域记录管理器消费， 正常注销路径的断连终态经攒批合并落库；编排层未构造批处理器时返回 nil）

func (*Hub) GetEphemeralCoalescer 

func (h *Hub) GetEphemeralCoalescer() *overload.Coalescer

GetEphemeralCoalescer 高频消息合并器（未启用时返回 nil）

func (*Hub) GetErrorStatsBatcher 

func (h *Hub) GetErrorStatsBatcher() *batcher.ErrorStatsBatcher

GetErrorStatsBatcher 连接错误统计批量聚合器

func (*Hub) GetGroupDisbandCallback 

func (h *Hub) GetGroupDisbandCallback() func(ctx context.Context, namespace, groupID string)

GetGroupDisbandCallback 群组解散回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetGroupMemberJoinCallback 

func (h *Hub) GetGroupMemberJoinCallback() func(ctx context.Context, namespace, groupID string, userIDs []string)

GetGroupMemberJoinCallback 群组成员加入回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetGroupMemberLeaveCallback 

func (h *Hub) GetGroupMemberLeaveCallback() func(ctx context.Context, namespace, groupID string, userIDs []string)

GetGroupMemberLeaveCallback 群组成员离开回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetGroupRepo 

func (h *Hub) GetGroupRepo() spi.GroupStore

GetGroupRepo 群组仓储（messaging.Host 端口名）

func (*Hub) GetGroupStore 

func (h *Hub) GetGroupStore() spi.GroupStore

GetGroupStore 群组仓储（group.Host 端口名）

func (*Hub) GetHeartbeatBatcher 

func (h *Hub) GetHeartbeatBatcher() *batcher.HeartbeatStatsUpdater

GetHeartbeatBatcher 心跳统计批量聚合器

func (*Hub) GetHeartbeatReportCallback 

func (h *Hub) GetHeartbeatReportCallback() func(client *models.Client)

GetHeartbeatReportCallback 心跳上报回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetHeartbeatTimeoutCallback 

func (h *Hub) GetHeartbeatTimeoutCallback() func(clientID string, userID string, lastHeartbeat time.Time)

GetHeartbeatTimeoutCallback 心跳超时回调（nil 表示业务方未注册，调用方须跳过）

func (*Hub) GetLogger 

func (h *Hub) GetLogger() spi.Logger

GetLogger 日志器

func (*Hub) GetMessageAggregator 

func (h *Hub) GetMessageAggregator() *overload.MessageAggregator

GetMessageAggregator 消息聚合器（未启用时返回 nil）

func (*Hub) GetMessageRecordOutbox 

func (h *Hub) GetMessageRecordOutbox() *batcher.MessageRecordOutbox

GetMessageRecordOutbox 消息记录攒批 outbox（write-ahead INSERT 攒批化）

func (*Hub) GetMessageRecordRepo 

func (h *Hub) GetMessageRecordRepo() spi.MessageSink

GetMessageRecordRepo 消息记录仓储（batcher.StorageBatchWriter 端口名）

func (*Hub) GetMessageSink 

func (h *Hub) GetMessageSink() spi.MessageSink

GetMessageSink 消息记录仓储

func (*Hub) GetMessageStatsBatcher 

func (h *Hub) GetMessageStatsBatcher() *batcher.MessageStatsBatcher

GetMessageStatsBatcher 消息统计批量聚合器

func (*Hub) GetMessageStatusUpdater 

func (h *Hub) GetMessageStatusUpdater() *batcher.MessageStatusUpdater

GetMessageStatusUpdater 消息状态批量更新器

func (*Hub) GetNodeID 

func (h *Hub) GetNodeID() string

GetNodeID 当前节点 ID

func (*Hub) GetObserverNotifier 

func (h *Hub) GetObserverNotifier() *batcher.ObserverNotificationBatcher

GetObserverNotifier 观察者通知批量处理器（messaging.Host 端口， 消息域 NotifyObservers 攒批入口经此提交）

func (*Hub) GetOnlineStatusRepo 

func (h *Hub) GetOnlineStatusRepo() spi.OnlineStore

GetOnlineStatusRepo 在线状态仓储

func (*Hub) GetOnlineUsersByType 

func (h *Hub) GetOnlineUsersByType(userType models.UserType) ([]string, error)

GetOnlineUsersByType 按用户类型查询在线用户（负载分配需要在线客服列表）

func (*Hub) GetOverloadMetrics 

func (h *Hub) GetOverloadMetrics() *overload.OverloadMetrics

GetOverloadMetrics 过载漏斗指标（编排层恒注入）

func (*Hub) GetShardedRegistry 

func (h *Hub) GetShardedRegistry() *connection.ShardedRegistry

GetShardedRegistry 分片连接注册表（核心运行时结构，恒注入）

func (*Hub) GetStartTime 

func (h *Hub) GetStartTime() time.Time

GetStartTime Hub 启动时间

func (*Hub) GetStatsRepo 

func (h *Hub) GetStatsRepo() spi.HubStats

GetStatsRepo 节点统计仓储（兼作消息计数开关）

func (*Hub) GetWorkloadStore 

func (h *Hub) GetWorkloadStore() spi.WorkloadStore

GetWorkloadStore 客服负载存储（未注入时调用方须报错而非 no-op）

func (*Hub) Group 

func (h *Hub) Group() *group.Manager

Group 群组域管理器

func (*Hub) HandleHeartbeat 

func (h *Hub) HandleHeartbeat(client *models.Client)

HandleHeartbeat 处理心跳消息（messaging.Host 端口委托） 协议级 PING 与应用层心跳共用同一保活入口，逻辑在连接域心跳管理器

func (*Hub) HasPubsub 

func (h *Hub) HasPubsub() bool

HasPubsub 是否配置分布式发布订阅（与 gRPC 共同构成跨节点通道开关）

func (*Hub) InitNodeGRPC 

func (h *Hub) InitNodeGRPC()

InitNodeGRPC 初始化节点间 gRPC 通信组件

在 SetPubSub 之后调用：节点发现依赖 Redis（从 PubSub 获取客户端） 若未启用 node-grpc 配置或 PubSub 未设置，则跳过初始化，Hub 退化为 Redis PubSub 模式

func (*Hub) IsGRPCEnabled 

func (h *Hub) IsGRPCEnabled() bool

IsGRPCEnabled 集群 gRPC 通道是否启用（配置开关判空，节点注册表由集群域初始化）

func (*Hub) IsShutdown 

func (h *Hub) IsShutdown() bool

IsShutdown 编排层是否正在关闭（transport.Registrar 端口，关闭中拒绝新连接）

func (*Hub) IsShuttingDown 

func (h *Hub) IsShuttingDown() bool

IsShuttingDown Hub 是否正在关闭（读泵据此区分服务端主动断开与异常断开）

func (*Hub) IsStarted 

func (h *Hub) IsStarted() bool

IsStarted Hub 是否已启动

func (*Hub) KickUser 

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

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

本地：踢出本节点上该用户信封内全部连接；sendNotification=true 时断链前向全部 连接写入 KickOut 通知（Guaranteed 级控制消息，notificationMsg 为通知文案）

幂等语义：用户已无连接（收集数 0）即"已离线"目标达成，KickedConnections=0 不视为失败，调用方据此区分"真踢到"与"本来就不在线"

跨节点：经在线路由索引查询用户连接所在的远端节点，异步分发 kick 指令 （gRPC 直连优先、PubSub 兜底）；远端节点按同一信封隔离静默踢出 （kick 指令不携带通知文案，通知仅由本节点发出）

func (*Hub) MarkRerouteAttempted 

func (h *Hub) MarkRerouteAttempted(messageID string, targetNodes []string, p2p bool)

MarkRerouteAttempted 标记消息已尝试重路由（防循环投递）

func (*Hub) Messaging 

func (h *Hub) Messaging() *messaging.Manager

Messaging 消息域管理器（发送 / 广播 / 分发 / ACK / 离线转存）

func (*Hub) NodeID 

func (h *Hub) NodeID() string

NodeID 节点 ID（K8s 兼容生成，见 hub.go 的 generateNodeID）

func (*Hub) NodeInfo 

func (h *Hub) NodeInfo() *models.NodeInfo

NodeInfo 节点信息（集群注册与路由用）

func (*Hub) NotifyObserverClients 

func (h *Hub) NotifyObserverClients(ctx context.Context, observers []*models.Client, msg *models.HubMessage) int32

NotifyObserverClients 向观察者客户端批量投递通知（观察者投递统一实现， messaging 本地路径 / PubSub 接收端 / gRPC 接收端三处共用，见 distributed.go）

func (*Hub) NotifyObservers 

func (h *Hub) NotifyObservers(ctx context.Context, msg *models.HubMessage)

NotifyObservers 通知观察者（观察者未启用时为 no-op） 从 ctx 提取 namespace+groupIDs 定位观察范围，提交批量处理器攒批投递

func (*Hub) NotifyObserversDirect 

func (h *Hub) NotifyObserversDirect(msg *models.HubMessage, namespace string, groupIDs []string)

NotifyObserversDirect 直接通知观察者（不经批处理队列，避免递归入队） 由 observerBatcher flush 调用：本地观察者投递 + 跨节点广播，无 per-message goroutine

func (*Hub) OnSlowConsumerEvicted 

func (h *Hub) OnSlowConsumerEvicted(client *models.Client, _ int, ratio float64)

OnSlowConsumerEvicted 实现 connection.EvictHook：慢消费者驱逐动作

扫描器已完成消息保全（SendChan 残留移交 ACK 超时链路兜底）与告警日志， 此处只做编排层三件事：

  1. 过载漏斗埋点（slow_evict）
  2. KickOut 控制消息通知客户端驱逐理由（控制通道独立 lane，不被业务洪峰淹没）
  3. Unregister 断链（断链不丢消息）

func (*Hub) OnWriteBatch 

func (h *Hub) OnWriteBatch(n int)

OnWriteBatch 写泵批量埋点（n = 本批消息数，闸门未启用时仅记漏斗计数）

func (*Hub) PendingMessages 

func (h *Hub) PendingMessages() int64

PendingMessages 返回尚未完成写出或断连清理的 WS 数据帧数量

func (*Hub) PublishGroupInvalidations 

func (h *Hub) PublishGroupInvalidations(ctx context.Context, appID string, groupIDs []string) error

PublishGroupInvalidations 实现群组域 Host 端口：批量发布群拓扑失效通知（聚合器的窗口出口） 复用全局广播频道（订阅侧自带 NodeID 自忽略去重）；单机形态 pubsub 未部署时短路返回 nil

func (*Hub) Register 

func (h *Hub) Register(client *models.Client)

Register 异步注册客户端（WS 升级路径：升级后立即返回，注册在后台完成） client.Context 在 http_upgrade 升级时已注入 trace_id，内部直接用实现全链路追踪

func (*Hub) RegisterSync 

func (h *Hub) RegisterSync(client *models.Client)

RegisterSync 同步注册客户端（SSE 路径：注册完成才进写循环，避免首条消息竞态丢失）

func (*Hub) Registry 

func (h *Hub) Registry() *connection.ShardedRegistry

Registry 分片连接注册表

func (*Hub) ReleaseDistributedLock 

func (h *Hub) ReleaseDistributedLock(ctx context.Context, key string) error

ReleaseDistributedLock 释放分布式锁

func (*Hub) RouteToCluster 

func (h *Hub) RouteToCluster(ctx context.Context, msg *models.HubMessage, opts cluster.ClusterDispatchOptions) error

RouteToCluster 路由消息到集群其他节点

func (*Hub) Run 

func (h *Hub) Run()

Run 启动Hub

func (*Hub) SafeShutdown 

func (h *Hub) SafeShutdown() error

SafeShutdown 安全关闭Hub，确保所有操作完成

func (*Hub) SendConditional 

func (h *Hub) SendConditional(ctx context.Context, condition func(*models.Client) bool, msg *models.HubMessage) int

SendConditional 条件广播：对满足条件的在线客户端各投递一份，返回投递数

func (*Hub) SendRegisteredMessage 

func (h *Hub) SendRegisteredMessage(client *models.Client)

SendRegisteredMessage 发送注册成功确认消息（配置启用时由传输域调用） 经消息域投递，走客户端写泵统一写出（单写者模式）

func (*Hub) SendToClient 

func (h *Hub) SendToClient(ctx context.Context, client *models.Client, msg *models.HubMessage)

SendToClient 定向单发（connection.LifecycleHost 端口，委托消息域） 踢出通知 / 注册确认等系统消息与业务消息共用同一投递路径

func (*Hub) SendToUserViaSSE 

func (h *Hub) SendToUserViaSSE(userID string, msg *models.HubMessage) bool

SendToUserViaSSE 经 SSE 通道向用户投递（SSE 未启用或用户无订阅时返回 false） namespace 隔离：msg.Namespace 非空时仅投递给同 ns 的 SSE 设备，避免跨 ns 串扰

func (*Hub) SendToUserWithRetry 

func (h *Hub) SendToUserWithRetry(ctx context.Context, toUserID string, msg *models.HubMessage) *models.SendResult

SendToUserWithRetry 点对点发送（带重试与路由注入），失败详情在返回值中

func (*Hub) SetAfterHeartbeatCallback 

func (h *Hub) SetAfterHeartbeatCallback(cb AfterHeartbeatCallback) *Hub

SetAfterHeartbeatCallback 设置心跳处理后回调（运行期可原子替换）

func (*Hub) SetBeforeHeartbeatCallback 

func (h *Hub) SetBeforeHeartbeatCallback(cb BeforeHeartbeatCallback) *Hub

SetBeforeHeartbeatCallback 设置心跳处理前回调（运行期可原子替换）

func (*Hub) SetClientConnectCallback 

func (h *Hub) SetClientConnectCallback(cb ClientConnectCallback) *Hub

SetClientConnectCallback 设置客户端连接回调（运行期可原子替换）

func (*Hub) SetClientDisconnectCallback 

func (h *Hub) SetClientDisconnectCallback(cb ClientDisconnectCallback) *Hub

SetClientDisconnectCallback 设置客户端断开回调（运行期可原子替换）

func (*Hub) SetConnectionQualityRepository 

func (h *Hub) SetConnectionQualityRepository(store spi.ConnectionQualityStore)

SetConnectionQualityRepository 注入连接质量仓储（StoreTarget 能力面）

func (*Hub) SetConnectionRecordRepository 

func (h *Hub) SetConnectionRecordRepository(store spi.ConnectionStore)

SetConnectionRecordRepository 注入连接记录仓储（StoreTarget 能力面）

func (*Hub) SetErrorCallback 

func (h *Hub) SetErrorCallback(cb messaging.ErrorCallback) *Hub

SetErrorCallback 设置统一错误处理回调（运行期可替换，委托消息域）

func (*Hub) SetGroupDisbandCallback 

func (h *Hub) SetGroupDisbandCallback(cb func(ctx context.Context, namespace, groupID string)) *Hub

SetGroupDisbandCallback 设置群组解散回调（运行期可原子替换）

func (*Hub) SetGroupMemberJoinCallback 

func (h *Hub) SetGroupMemberJoinCallback(cb func(ctx context.Context, namespace, groupID string, userIDs []string)) *Hub

SetGroupMemberJoinCallback 设置群组成员加入回调（运行期可原子替换）

func (*Hub) SetGroupMemberLeaveCallback 

func (h *Hub) SetGroupMemberLeaveCallback(cb func(ctx context.Context, namespace, groupID string, userIDs []string)) *Hub

SetGroupMemberLeaveCallback 设置群组成员离开回调（运行期可原子替换）

func (*Hub) SetGroupRepository 

func (h *Hub) SetGroupRepository(store spi.GroupStore)

SetGroupRepository 注入群组仓储（StoreTarget 能力面） 统一注入拓扑失效回调（GroupStore 契约能力）：持本地缓存的实现（GroupMemberCache） 写路径本地逐出后回调群组域聚合器标记 dirty，100ms 窗口批量广播，跨节点写一致性 窗口 30s → ~100ms；无缓存的裸仓储 no-op 丢弃

func (*Hub) SetHeartbeatReportCallback 

func (h *Hub) SetHeartbeatReportCallback(cb HeartbeatReportCallback) *Hub

SetHeartbeatReportCallback 设置心跳上报回调（运行期可原子替换）

func (*Hub) SetHeartbeatTimeoutCallback 

func (h *Hub) SetHeartbeatTimeoutCallback(cb HeartbeatTimeoutCallback) *Hub

SetHeartbeatTimeoutCallback 设置心跳超时回调（运行期可原子替换）

func (*Hub) SetHubStatsRepository 

func (h *Hub) SetHubStatsRepository(store spi.HubStats)

SetHubStatsRepository 注入 Hub 统计仓储（StoreTarget 能力面）

func (*Hub) SetMessageReceivedCallback 

func (h *Hub) SetMessageReceivedCallback(cb messaging.MessageReceivedCallback) *Hub

SetMessageReceivedCallback 设置客户端上行消息回调（运行期可替换，委托消息域）

func (*Hub) SetMessageRecordRepository 

func (h *Hub) SetMessageRecordRepository(sink spi.MessageSink)

SetMessageRecordRepository 注入消息记录仓储（StoreTarget 能力面）

func (*Hub) SetMessageSendCallback 

func (h *Hub) SetMessageSendCallback(cb messaging.MessageSendCallback) *Hub

SetMessageSendCallback 设置消息发送完成回调（运行期可替换，委托消息域）

func (*Hub) SetOfflineMessageHandler 

func (h *Hub) SetOfflineMessageHandler(queue spi.OfflineQueue)

SetOfflineMessageHandler 注入离线消息处理器（StoreTarget 能力面，直通 messagingMgr）

处理器实现 spi.OfflineQueue 契约，持久化全部经注入的共享存储 （Redis 队列 + RDBMS，Deployment 滚动更新下跨 Pod 可见，Pod 本地无状态）； 便捷组装见 messaging.InitializeOfflineQueue

func (*Hub) SetOfflineMessagePushCallback 

func (h *Hub) SetOfflineMessagePushCallback(cb OfflineMessagePushCallback) *Hub

SetOfflineMessagePushCallback 设置离线消息推送回调（运行期可原子替换）

func (*Hub) SetOnlineStatusRepository 

func (h *Hub) SetOnlineStatusRepository(store spi.OnlineStore)

SetOnlineStatusRepository 注入在线状态仓储（StoreTarget 能力面）

func (*Hub) SetOverloadPolicy 

func (h *Hub) SetOverloadPolicy(gate *overload.AdmissionGate, shaper *overload.Shaper, coalescer *overload.Coalescer, aggregator *overload.MessageAggregator) *Hub

SetOverloadPolicy 链式配置过载保护策略（运行期热替换）

参数为 nil 表示保留该组件现状（不替换）；闸门替换时旧实例 Stop、 新实例 Store 后立即 Start —— 已 Run 的 Hub 调用时新闸门直接进入 水位评估循环（atomic Store 发布：后台循环并发 Load 读到完整初始化的实例）。 组件构造见 overload.NewAdmissionGate / NewShaper / NewCoalescer / NewMessageAggregator （零值参数由 constants 包默认值兜底）

func (*Hub) SetWorkloadRepository 

func (h *Hub) SetWorkloadRepository(store spi.WorkloadStore)

SetWorkloadRepository 注入客服负载仓储（StoreTarget 能力面）

func (*Hub) Shutdown 

func (h *Hub) Shutdown()

Shutdown 关闭Hub（旧API，兼容性方法）

func (*Hub) Stats 

func (h *Hub) Stats() *stats.Manager

Stats 统计域管理器

func (*Hub) SubmitClusterDispatch 

func (h *Hub) SubmitClusterDispatch(msg *models.HubMessage, opts cluster.ClusterDispatchOptions) bool

SubmitClusterDispatch 提交集群批量分发任务到分布式池（队列满时返回 false） 分发沿用消息路由信封上下文，保留 trace_id 实现全链路追踪

func (*Hub) SubscribeBroadcastChannel 

func (h *Hub) SubscribeBroadcastChannel(ctx context.Context) error

SubscribeBroadcastChannel 订阅全局广播频道

func (*Hub) SubscribeNodeMessages 

func (h *Hub) SubscribeNodeMessages(ctx context.Context) error

SubscribeNodeMessages 订阅本节点的消息通道

func (*Hub) SubscribeObserverChannel 

func (h *Hub) SubscribeObserverChannel(ctx context.Context) error

SubscribeObserverChannel 订阅观察者通知频道

func (*Hub) TrackConnectionError 

func (h *Hub) TrackConnectionError(connectionID string, userType models.UserType, err error)

TrackConnectionError 记录连接错误（异常断开排查）

func (*Hub) TrackHeartbeatStats 

func (h *Hub) TrackHeartbeatStats(client *models.Client)

TrackHeartbeatStats 投递心跳统计（connection.HeartbeatHost 端口，stats 域漏斗）

func (*Hub) TrackReceiverMessageStats 

func (h *Hub) TrackReceiverMessageStats(connectionID string, receiverType models.UserType, dataSize int)

TrackReceiverMessageStats 投递接收方消息统计

func (*Hub) TrySubmitCallback 

func (h *Hub) TrySubmitCallback(task func()) bool

TrySubmitCallback 提交一个异步回调任务（队列满时返回 false） 群组生命周期回调走此处，避免每条连接一个 goroutine

func (*Hub) Unregister 

func (h *Hub) Unregister(client *models.Client)

Unregister 异步注销客户端（写循环退出后的兜底清理，幂等）

func (*Hub) Wait 

func (h *Hub) Wait()

Wait 等待所有显式登记进 h.wg 的后台 goroutine 退出（SafeShutdown 完成后返回） 当前登记项：Run 事件循环、心跳 Redis worker（processHeartbeatRedisUpdates）、 注册/注销异步任务（Register/Unregister 的 go 协程） 注意：syncx.Go 启动的订阅/drain/扫描协程未绑定 h.wg，依赖 h.ctx 取消退出； 消息域读写泵由 messaging.Manager.Wait 单独等待（不同接收者，互不冲突）

func (*Hub) WaitForStart 

func (h *Hub) WaitForStart()

WaitForStart 等待Hub启动完成 这个方法对于用户来说很重要，确保Hub完全启动后再进行操作

func (*Hub) WaitForStartWithTimeout 

func (h *Hub) WaitForStartWithTimeout(timeout time.Duration) error

WaitForStartWithTimeout 带超时的等待Hub启动

func (*Hub) WithAfterHeartbeatCallback 

func (h *Hub) WithAfterHeartbeatCallback(cb AfterHeartbeatCallback) *Hub

WithAfterHeartbeatCallback 注入心跳处理后回调

func (*Hub) WithBeforeHeartbeatCallback 

func (h *Hub) WithBeforeHeartbeatCallback(cb BeforeHeartbeatCallback) *Hub

WithBeforeHeartbeatCallback 注入心跳处理前回调（返回 false 跳过后续心跳处理）

func (*Hub) WithClientConnectCallback 

func (h *Hub) WithClientConnectCallback(cb ClientConnectCallback) *Hub

WithClientConnectCallback 注入客户端连接回调（权限验证 / 会话初始化等）

func (*Hub) WithClientDisconnectCallback 

func (h *Hub) WithClientDisconnectCallback(cb ClientDisconnectCallback) *Hub

WithClientDisconnectCallback 注入客户端断开回调（资源清理 / 在线状态更新等）

func (*Hub) WithConnectionQualityStore 

func (h *Hub) WithConnectionQualityStore(store spi.ConnectionQualityStore) *Hub

WithConnectionQualityStore 注入连接质量仓储（MySQL；须在 Run() 前注入）

func (*Hub) WithConnectionStore 

func (h *Hub) WithConnectionStore(store spi.ConnectionStore) *Hub

WithConnectionStore 注入连接记录仓储（MySQL；须在 Run() 前注入）

func (*Hub) WithConnectionTokenDecoder 

func (h *Hub) WithConnectionTokenDecoder(decoder spi.ConnectionAuthenticator) *Hub

WithConnectionTokenDecoder 注入连接 Token 鉴权器（nil 时握手走明文参数）

func (*Hub) WithErrorCallback 

func (h *Hub) WithErrorCallback(cb messaging.ErrorCallback) *Hub

WithErrorCallback 注入统一错误处理回调（日志记录 / 告警通知等，委托消息域）

func (*Hub) WithEventBus 

func (h *Hub) WithEventBus(bus spi.EventBus) *Hub

WithEventBus 注入事件总线并装配消费域（须在 Run() 前注入）

bus 非空即 MQ 模式——Run() 时订阅 deliver/broadcast 双通道， 消费收敛到 hub.Deliver 决策树（零新增投递逻辑）； 未注入即直调模式，broker 域零初始化零开销。去重窗参数走 broker 包内置默认 （等级分流去重不开放配置，避免死配置）

func (*Hub) WithGRPCClientPool 

func (h *Hub) WithGRPCClientPool(pool *cluster.GRPCClientPool) *Hub

WithGRPCClientPool 注入节点间 gRPC 客户端连接池

func (*Hub) WithGroupDisbandCallback 

func (h *Hub) WithGroupDisbandCallback(cb func(ctx context.Context, namespace, groupID string)) *Hub

WithGroupDisbandCallback 注入群组解散回调（DisbandGroup 成功后异步触发）

func (*Hub) WithGroupMemberJoinCallback 

func (h *Hub) WithGroupMemberJoinCallback(cb func(ctx context.Context, namespace, groupID string, userIDs []string)) *Hub

WithGroupMemberJoinCallback 注入群组成员加入回调（连接时自动加群成功后异步触发）

func (*Hub) WithGroupMemberLeaveCallback 

func (h *Hub) WithGroupMemberLeaveCallback(cb func(ctx context.Context, namespace, groupID string, userIDs []string)) *Hub

WithGroupMemberLeaveCallback 注入群组成员离开回调（RemoveGroupMembers 成功后异步触发）

func (*Hub) WithGroupStore 

func (h *Hub) WithGroupStore(store spi.GroupStore) *Hub

WithGroupStore 注入群组仓储（Redis；须在 Run() 前注入）

func (*Hub) WithHeartbeatReportCallback 

func (h *Hub) WithHeartbeatReportCallback(cb HeartbeatReportCallback) *Hub

WithHeartbeatReportCallback 注入心跳上报回调

func (*Hub) WithHeartbeatTimeoutCallback 

func (h *Hub) WithHeartbeatTimeoutCallback(cb HeartbeatTimeoutCallback) *Hub

WithHeartbeatTimeoutCallback 注入心跳超时回调

func (*Hub) WithLogger 

func (h *Hub) WithLogger(logger spi.Logger) *Hub

WithLogger 注入日志器（NewHub 已用 spi.InitLogger 构造默认值）

func (*Hub) WithMessageReceivedCallback 

func (h *Hub) WithMessageReceivedCallback(cb messaging.MessageReceivedCallback) *Hub

WithMessageReceivedCallback 注入客户端上行消息回调（业务逻辑处理 / 消息路由，委托消息域）

func (*Hub) WithMessageSendCallback 

func (h *Hub) WithMessageSendCallback(cb messaging.MessageSendCallback) *Hub

WithMessageSendCallback 注入消息发送完成回调（含重试信息与最终错误，委托消息域）

func (*Hub) WithMessageSink 

func (h *Hub) WithMessageSink(sink spi.MessageSink) *Hub

WithMessageSink 注入消息发送记录仓储（MySQL 归档 + 统计；须在 Run() 前注入， Run 后注入不会启动记录清理等 IfTicker 且存在数据竞争）

注意：离线消息处理器与本仓储无关 —— 经 WithOfflineMessageHandler 注入， 或由 messaging.InitializeOfflineQueue 经 spi.StoreTarget 的 SetOfflineMessageHandler 装配（见 interfaces.go 的 StoreTarget 区块）。

func (*Hub) WithNodeRegistry 

func (h *Hub) WithNodeRegistry(registry *cluster.NodeRegistry) *Hub

WithNodeRegistry 注入节点注册表（集群成员发现与心跳）

func (*Hub) WithOfflineMessageHandler 

func (h *Hub) WithOfflineMessageHandler(handler spi.OfflineQueue) *Hub

WithOfflineMessageHandler 注入离线消息处理器（须在 Run() 前注入）

处理器实现 spi.OfflineQueue 契约（混合实现见 messaging.HybridOfflineMessageHandler： Redis 队列 + RDBMS 双写），注入后挂接到 messagingMgr —— 投递失败转存、用户全端 离线预存等消费链路随即生效。持久化全部经注入的共享存储，Pod 本地无状态； 两半依赖的便捷组装见 messaging.InitializeOfflineQueue（业务侧亦可直接传构造好的 handler）

func (*Hub) WithOfflineMessagePushCallback 

func (h *Hub) WithOfflineMessagePushCallback(cb OfflineMessagePushCallback) *Hub

WithOfflineMessagePushCallback 注入离线消息推送完成回调（上游据此删除已推送消息）

func (*Hub) WithOnlineStatusRepo 

func (h *Hub) WithOnlineStatusRepo(repo spi.OnlineStore) *Hub

WithOnlineStatusRepo 注入在线状态仓储（Redis）

须在 Run() 前注入：Run 启动时按该字段是否非空决定是否拉起心跳 Redis worker （processHeartbeatRedisUpdates），Run 后注入不会启动该 worker 且存在数据竞争

func (*Hub) WithPubsub 

func (h *Hub) WithPubsub(pubsub *cachex.PubSub) *Hub

WithPubsub 注入发布订阅（跨节点事件通道，单机模式可缺省）

func (*Hub) WithStatsRepo 

func (h *Hub) WithStatsRepo(repo spi.HubStats) *Hub

WithStatsRepo 注入 Hub 统计仓储（Redis；须在 Run() 前注入）

func (*Hub) WithWelcomeProvider 

func (h *Hub) WithWelcomeProvider(provider models.WelcomeMessageProvider) *Hub

WithWelcomeProvider 注入欢迎消息提供者

func (*Hub) WithWorkloadStore 

func (h *Hub) WithWorkloadStore(store spi.WorkloadStore) *Hub

WithWorkloadStore 注入客服负载管理仓储（Redis + MySQL；须在 Run() 前注入）

type HubCallbacks 

type HubCallbacks struct {
	// 连接生命周期
	ClientConnect    ClientConnectCallback
	ClientDisconnect ClientDisconnectCallback

	// 离线推送完成（上游据此删除已推送消息）
	OfflineMessagePush OfflineMessagePushCallback

	// 心跳生命周期（BeforeHeartbeat 返回 false 跳过后续心跳处理）
	HeartbeatTimeout HeartbeatTimeoutCallback
	HeartbeatReport  HeartbeatReportCallback
	BeforeHeartbeat  BeforeHeartbeatCallback
	AfterHeartbeat   AfterHeartbeatCallback

	// 群组生命周期（结构类型与 group.Host 端口一致）
	GroupDisband     func(ctx context.Context, namespace, groupID string)
	GroupMemberJoin  func(ctx context.Context, namespace, groupID string, userIDs []string)
	GroupMemberLeave func(ctx context.Context, namespace, groupID string, userIDs []string)
}

HubCallbacks 应用层回调集合（连接/心跳/离线推送/群组生命周期）

回调未注入（nil）时对应环节直接跳过。运行期热替换经 Hub.callbacks 的 atomic.Pointer 整体快照交换（replaceCallback）完成——Go 函数值是双字 （type word + data word），裸字段并发读写可撕裂出损坏的函数指针， 读方（注册/心跳/群组路径）高频并发读，必须走 snapshotCallbacks 快照

type OfflineMessagePushCallback 

type OfflineMessagePushCallback func(userID string, pushedMessageIDs []string, failedMessageIDs []string)

OfflineMessagePushCallback 离线消息推送回调

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.