messaging

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

README

messaging（消息域）

收敛 → 投递 → ACK → 离线的完整闭环——每一级推送都有明确收敛路径，绝不全量扫描

这个域干什么

messaging 是消息的进出总管：入站解码分发、路由收敛、P2P / 广播投递、ACK 确认、 失败转离线、重试与查询、消息记录

v4 起本域吸收三个旧包（裁决记录 #10）：

  • inbound/（dispatcher / mux / forward / probe）→ 入站半边
  • offline/（HybridOfflineMessageHandler）→ 投递失败出口
  • protocol/（AckManager / AckStatus / PendingMessage）→ 协议半边

核心概念

推送三级 API（路由收敛模型，REFACTORING_DESIGN §五.5）
级别 API 收敛路径
P2P Send(ctx, msg, targets...) 本地 registry 命中直投；未命中查路由缓存跨节点
定向广播 BroadcastApps / BroadcastNamespaces / BroadcastGroups appID → namespace → groupID → userID 并集，层层索引收敛
全局广播 Broadcast(msg) RangeParallel 64 shard 并行扇出 + 合批 writev

收敛铁律：定向广播在源节点完成收敛，远端收到的是 userID 列表，只做 P2P—— 索引在内存（group 域三层订阅索引），每层收敛是一次查找而非全量过滤

其余核心概念
概念 说明
dispatcher / mux 入站解码与两级路由（MessageType → Topic）
观察者投递 NotifyObservers 攒批入口 → batcher 合并；NotifyObserversDirect 直投（三级索引查找 + 预序列化共享 + 跨节点广播），见 observer.go
SSE 投递 SendToUserViaSSE 点对点（O(1) 用户索引 + ns 隔离）；BroadcastToSSEClients 全量广播（并行分片 + 信封隔离），见 sse.go
上下文键 ContextKeySenderID / ContextKeyUserID（连接归属，transport 注入）/ ContextKeyOfflineBroadcastCollector（扇出聚合器，context_keys.go）
AckManager 发送 → 时间轮插入 (msgID, deadline)；收到 ACK O(1) 取消；超时标记 AckTimeout
ackTimeoutSlack 跨节点 ACK 的抖动余量，防误判超时
HybridOfflineMessageHandler Redis + 本地双路径离线队列（按 P2P / group 维度一致性 key）
DeliveryGuarantee 消息级保证：at-least-once / exactly-once（单节点内 Outbox）
worker_pool 分池写，单连接单写 goroutine，天然防交叉写
投递结果与过载积压统计
  • 积压口径：WS 积压 = 经 Client.TrySend 接纳、尚未完成写出或清理的数据帧数；P2P、广播、观察者共用此入口，入队失败回滚、写出结算、断连释放，控制通道与 SSE 不混入统计。Hub 挂独立实时计数（替换准入闸门时保留已有积压），Hub.PendingMessages 即真实待写帧数
  • 投递成功语义：至少一个匹配客户端接纳才算在线投递成功（队列接纳 ≠ 客户端 ACK）；目标节点用户在线但拒绝投递返回 ResourceExhausted，单条与微批 gRPC 均保留该分类——不转成用户不在线、不清理在线索引、不降级 PubSub 后误报成功
  • 离线兜底：P2P 最终兜底同步确认存储结果后才返回 StoredOffline=true；回放暂存消息补投失败同样接回离线兜底；notice 支持离线重放，其余被策略排除的系统消息返回 spi.ErrOfflineStorageSkipped（跳过 ≠ 保存成功）

怎么用

package main

import (
    "context"

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

h := hub.NewHub(cfg)
ctx := routing.NewRoute().WithAppID("app-1").WithNamespace("ns-1").Inject(ctx)

// P2P
h.Send(ctx, msg, "user-1001")

// 定向广播：层层收敛（appID+namespace+多 groupID）
h.BroadcastGroups(ctx, msg, "app-1", "ns-1", "group-a", "group-b")

// 全局广播（全 appID 收）
h.Broadcast(ctx, msg)

端口与依赖

  • 消费 group 域的三层订阅索引（收敛数据源）
  • 消费 cluster 域的跨节点投递（收敛后的远端 P2P）
  • 依赖 spi 契约：OfflineQueue（离线）、MessageSink（记录）、ArchiveSink（归档）

性能与陷阱

  • P4：逐帧 writev → N 次 syscall；必须合批 + sync.Pool 帧复用
  • P16：定向广播全量过滤 → 退化为 O(全连接)；索引必须在内存，Redis 只做跨节点兜底
  • ACK 时间轮精度要大于跨节点 RTT 抖动（node_ack_timeout.go 兜底）
  • 多设备：收敛到 userID 后展开该 user 的全部活跃连接（user/device 双维度）

后端绑定

  • NATS：跨节点兜底投递通道（wsc.node.<nodeID>，JetStream 持久化重试）
  • Redis：离线队列（HybridOfflineMessageHandler）
  • 不启用时：单节点全内存，投递失败即转本地离线缓冲（可配丢弃）

Documentation

Overview

* @Author: kamalyes 501893067@qq.com * @Date: 2025-11-15 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\messaging\ack.go * @Description: ACK消息确认机制 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-22 10:08:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-22 10:08:00 * @FilePath: \go-wsc\messaging\context_keys.go * @Description: 消息域上下文键 —— 连接归属与离线广播聚合器的单一事实源 * * 迁移自 constants/context.go（P2 批2 域化）： * - SenderID/UserID 键由 transport 升级器在连接 ctx 注入、消息域消费， * 键与消费语义同属消息链路，归位消息域（transport 单向依赖 messaging，无循环） * - 离线广播聚合器键供批量扇出路径聚合未命中用户，扇出结束后统一跨节点推送 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-07 21:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-22 10:42:00 * @FilePath: \go-wsc\messaging\delivery_fallback.go * @Description: 投递兜底路由 —— 拒绝不等于丢弃（P0 分级送达基座） * * TrySend 返回 false（SendChan 满/客户端关闭）时按消息分级路由： * - 必达级（Guaranteed）：立即转离线补发（ACK 超时链路仍作双保险——时间轮到期后 * ClaimStaleSending 转离线，此处提前到"入队失败即转"，不等 5min 兜底） * - 普通级（Standard）：立即转离线（用户上线时推送——修复"广播丢弃彻底丢失"） * - 高频级（Ephemeral）：丢弃计数（latest-wins 语义：下一条同 key 消息自然覆盖， * 旧值过期是正确行为而非丢失） * * 广播路径的成员级兜底（BroadcastToFiltered/BroadcastToUserIDs 的 TrySend false 分支） * 复用同一路由：消息 Receiver 为空时补写目标用户再转离线，修复广播离线丢失缺陷 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-11 22:30:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-22 10:57:00 * @FilePath: \go-wsc\messaging\frame_writer.go * @Description: writev 合帧批量写 —— 突发 N 条消息 N 次 syscall 降至 1 次 writev * * RFC6455 帧自定界（服务端帧无掩码），net.Buffers 聚合多个 [帧头+载荷] * 一次 writev 落盘。首条消息直写（低延迟），积压部分合帧（吞吐）： * - 无积压：1 条 1 次 syscall（与原路径持平） * - 突发 N 条：1 + 1 次 syscall（首条直写 + N-1 合帧一次写出） * * 写锁安全性：写泵是唯一数据写者（单写者模型，与原 WriteMessage 路径一致）； * WriteControl（协议 ping/pong）与数据写的理论交错窗口与原路径相同（见 * closeClientConnection 注释——控制帧独立锁，交错时客户端走异常重连，不影响正确性） * 压缩扩展未启用（DefaultUpgrader 未开 EnableCompression），帧格式为标准无压缩帧 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-22 10:35:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-26 09:38:02 * @FilePath: \go-wsc\messaging\json_peek.go * @Description: JSON 顶层字段就地探测（入站控制类消息快路径，免完整反序列化） * * 动机：handleTextMessage 原先用一个 2 字段的探测结构体做 json.Unmarshal 来识别 * 心跳/ACK，命中即返回。但 json.Unmarshal 即使只映射 2 个字段也要扫描并跳过整个文档， * 于是**非控制类消息（数据消息，占数据面绝大多数）要付两次完整解析**： * 一次探测 + 一次真正的 HubMessage 反序列化。 * * 本文件提供只读的顶层字符串字段探测：单次线性扫描，零分配，不构造中间对象。 * * 正确性优先于速度——探测只在**无歧义**时返回命中： * - 只认深度 0（最外层对象）的键，嵌套对象里的同名键不算 * - 字符串内部的字节不参与键匹配（含转义处理），避免 content 里出现 * "message_type" 字样时误判 * - 任何结构异常（截断、非对象、非字符串值）一律返回未命中 * * 误判方向的安全性：返回"未命中"是安全的——调用方会落到完整反序列化， * 且完整解析后仍会复查 MessageType 作兜底（见 handleTextMessage 的类型 switch）。 * 返回"错误的命中"才是危险方向，故上述三条约束缺一不可。

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-23 10:05:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-23 10:05:00 * @FilePath: \go-wsc\messaging\replay_gate.go * @Description: 首连离线回放门闩（重连场景的投递顺序闭环） * * 问题背景：首连触发的 PushOfflineMessages 异步执行（workerPool 回调池）， * 期间新到的实时消息直接进 sendChan → 用户先收到新消息、后收到更早的 * 离线消息（乱序）；drain 队列还可能把重连瞬间被误转离线的消息重复推出 * * 解法：首连时 begin 门闩 → 该用户的实时投递被暂存 → 离线回放完成后 * end 按暂存顺序补投 → 开闸。每用户投递顺序 = 消息真实时序 * * 边界保护： * - 每用户暂存上限 maxReplayHoldout：超出后直接投递不再暂存（宁可极小 * 概率乱序，也不无限堆积内存） * - end 由 defer 保证必达：回放 panic 也会开闸，闸门不会永久卡死 * - 单机模式无离线处理器时不启用门闩（PushOfflineMessages nil-safe 跳过） * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func InitializeOfflineQueue 

func InitializeOfflineQueue(hub spi.StoreTarget, deps OfflineDeps) error

InitializeOfflineQueue 组装并注入混合离线消息处理器

两半依赖任一缺失时返回错误而非静默跳过：离线消息是「用户离线期间的 消息不丢」这条承诺的唯一兜底，缺一半会让消息在无人察觉的情况下丢失。 若业务侧确实不需要离线消息能力，应显式不调用本函数。

func NewReplayGate 

func NewReplayGate() *replayGate

NewReplayGate 创建门闩（编排层构造 messaging.Manager 时经 WithReplayGate 注入）

Types

type AckManager 

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

AckManager ACK管理器 基于 syncx.ShardedMap 实现待确认消息的分片存储，替代单一 mutex + map 高并发下 AddPending/Confirm/Remove 操作完全并行，锁竞争降低 64 倍

func NewAckManager 

func NewAckManager(ackTimeout time.Duration, maxRetry int) *AckManager

NewAckManager 创建ACK管理器

func NewAckManagerWithOptions 

func NewAckManagerWithOptions(ackTimeout time.Duration, maxRetry int, expireDuration time.Duration, offlineRepo spi.OfflineStore) *AckManager

NewAckManagerWithOptions 创建ACK管理器（带选项）

func (*AckManager) AddPendingMessage 

func (am *AckManager) AddPendingMessage(msg *models.HubMessage) *PendingMessage

AddPendingMessage 添加待确认消息

func (*AckManager) AddPendingMessageWithExpire 

func (am *AckManager) AddPendingMessageWithExpire(msg *models.HubMessage, timeout time.Duration, maxRetry int) *PendingMessage

AddPendingMessageWithExpire 添加待确认消息(带过期时间)

func (*AckManager) CleanupExpired 

func (am *AckManager) CleanupExpired() int

CleanupExpired 清理过期的待确认消息 使用 ShardedMap.Range 遍历（分片读锁粒度），收集过期 key 后批量删除

func (*AckManager) ConfirmMessage 

func (am *AckManager) ConfirmMessage(messageID string, ack *AckMessage) bool

ConfirmMessage 确认消息

func (*AckManager) GetMaxRetry 

func (am *AckManager) GetMaxRetry() int

GetMaxRetry 获取最大重试次数

func (*AckManager) GetPendingCount 

func (am *AckManager) GetPendingCount() int

GetPendingCount 获取待确认消息数量（原子读取，零锁开销）

func (*AckManager) GetPendingMessage 

func (am *AckManager) GetPendingMessage(messageID string) (*PendingMessage, bool)

GetPendingMessage 获取待确认消息

func (*AckManager) GetTimeout 

func (am *AckManager) GetTimeout() time.Duration

GetTimeout 获取ACK超时时间

func (*AckManager) RemovePendingMessage 

func (am *AckManager) RemovePendingMessage(messageID string)

RemovePendingMessage 移除待确认消息

func (*AckManager) SetExpireDuration 

func (am *AckManager) SetExpireDuration(duration time.Duration)

SetExpireDuration 设置消息过期时间

func (*AckManager) SetOfflineRepo 

func (am *AckManager) SetOfflineRepo(handler spi.OfflineStore)

SetOfflineRepo 设置离线消息处理器 用于统一使用 Hub 的离线消息处理器

func (*AckManager) Shutdown 

func (am *AckManager) Shutdown()

Shutdown 关闭ACK管理器 遍历所有待确认消息取消 context，然后清空分片存储

type AckMessage 

type AckMessage struct {
	MessageID string    `json:"message_id"` // HubMessage消息ID
	Status    AckStatus `json:"status"`     // ACK状态
	Timestamp time.Time `json:"timestamp"`  // 时间戳
	Error     string    `json:"error"`      // 错误信息
}

AckMessage ACK消息结构

type AckStatus 

type AckStatus string

AckStatus ACK状态 

const (
	AckStatusPending   AckStatus = "pending"   // 等待确认
	AckStatusConfirmed AckStatus = "confirmed" // 已确认
	AckStatusTimeout   AckStatus = "timeout"   // 超时
	AckStatusFailed    AckStatus = "failed"    // 失败
)

type ContextKey 

type ContextKey string

ContextKey 上下文键类型（消息域：连接归属用户 + 离线广播聚合器） 

const (
	// ContextKeyUserID 用户ID
	ContextKeyUserID ContextKey = "user_id"
	// ContextKeySenderID 发送者ID（连接级 ctx 注入，标识连接归属用户）
	ContextKeySenderID ContextKey = "sender_id"
	// ContextKeyOfflineBroadcastCollector 批量扇出路径的离线广播聚合器
	//
	// 群组成员可达数万，扇出 goroutine 内逐人同步广播会造成 N 倍集群流量放大与
	// Redis 连接池耗尽（OOM 教训）；但完全跳过又会丢失"索引滞后但实际在线"用户的实时投递。
	// 折中：扇出期间仅聚合收集未命中用户，扇出结束后统一逐人跨节点推送（真的推送），
	// 真正离线的用户由离线存储 + 重连上线拉取兜底
	ContextKeyOfflineBroadcastCollector ContextKey = "offline_broadcast_collector"
)

type ErrorCallback 

type ErrorCallback func(ctx context.Context, err error, severity models.ErrorSeverity) error

ErrorCallback 错误处理回调（统一处理消息链路各环节错误）

type Host 

type Host interface {

	// Context 返回 Hub 生命周期上下文
	Context() context.Context
	// GetLogger 日志器
	GetLogger() spi.Logger
	// GetNodeID 当前节点 ID
	GetNodeID() string
	// GetConfig 全局配置
	GetConfig() *wscconfig.WSC
	// IsShuttingDown Hub 是否正在关闭（读泵据此区分服务端主动断开与异常断开）
	IsShuttingDown() bool

	// GetShardedRegistry 分片连接注册表（未注入时返回 nil）
	GetShardedRegistry() *connection.ShardedRegistry
	// Unregister 注销客户端连接（连接生命周期归编排层）
	Unregister(client *models.Client)
	// HandleHeartbeat 处理心跳消息（连接域时间轮续期 + 统计刷新）
	HandleHeartbeat(client *models.Client)

	// GetMessageSink 消息记录仓储（未注入时返回 nil）
	GetMessageSink() spi.MessageSink
	// GetGroupRepo 群组仓储（未注入时返回 nil）
	GetGroupRepo() spi.GroupStore
	// GetStatsRepo 节点统计仓储（未注入时返回 nil，兼作消息计数开关）
	GetStatsRepo() spi.HubStats
	// HasPubsub 是否配置分布式发布订阅（与 gRPC 共同构成跨节点通道开关）
	HasPubsub() bool

	// IsGRPCEnabled 集群 gRPC 通道是否启用
	IsGRPCEnabled() bool
	// RouteToCluster 路由消息到集群其他节点
	RouteToCluster(ctx context.Context, msg *models.HubMessage, opts cluster.ClusterDispatchOptions) error
	// BatchGetUserNodes 批量查询多个用户所在节点（Pipeline 单次往返，信封过滤与路由上下文一致）
	// 单个用户查询传单元素切片复用本端口（不再保留单查方法）；
	// 在线判定语义：len(nodes)>0 即在线；查询失败/未注入返回 nil，调用方回退逐用户查询
	BatchGetUserNodes(ctx context.Context, userIDs []string) map[string][]string
	// CheckAndRouteToNode 检查用户在线节点并按需跨节点投递
	// presetNodes 非空时跳过节点查询（调用方已预取，消第二次 Redis 往返）
	// 返回：是否命中跨节点路由、目标节点列表、错误
	CheckAndRouteToNode(ctx context.Context, userID string, msg *models.HubMessage, presetNodes []string) (bool, []string, error)
	// GetAllClusterNodeIDs 获取集群全部节点 ID
	GetAllClusterNodeIDs() []string
	// MarkRerouteAttempted 标记消息已尝试重路由（防循环投递）
	MarkRerouteAttempted(messageID string, targetNodes []string, p2p bool)
	// DeleteRerouteGuard 清理重路由守卫条目（消息到达终态时调用）
	DeleteRerouteGuard(messageID string)
	// SubmitClusterDispatch 提交集群批量分发任务（队列满时返回 false）
	SubmitClusterDispatch(msg *models.HubMessage, opts cluster.ClusterDispatchOptions) bool

	// GetMessageRecordOutbox 消息记录攒批 outbox（未注入时返回 nil，消息域降级单条 Create）
	GetMessageRecordOutbox() *batcher.MessageRecordOutbox
	// TrackReceiverMessageStats 投递接收方消息统计
	TrackReceiverMessageStats(connectionID string, receiverType models.UserType, dataSize int)
	// TrackConnectionError 记录连接错误（异常断开排查）
	TrackConnectionError(connectionID string, userType models.UserType, err error)
	// GetObserverNotifier 观察者通知批量处理器（未注入时返回 nil，观察者通知降级 no-op）
	GetObserverNotifier() *batcher.ObserverNotificationBatcher
	// NotifyObserverClients 向观察者客户端批量投递通知（观察者投递统一实现：
	// Clone + observer metadata + 预序列化一次 + TrySend 直投，旁路投递不走状态回报）
	// 返回成功投递的观察者数
	NotifyObserverClients(ctx context.Context, observers []*models.Client, msg *models.HubMessage) int32

	// AdmitMessage 消息级准入裁决（闸门未启用时恒放行）
	AdmitMessage(msg *models.HubMessage, isBroadcast bool) overload.AdmitVerdict
	// AdmissionOnDelivered 投递完成埋点（闸门未启用时 no-op）
	AdmissionOnDelivered()
	// OnWriteBatch 写泵批量埋点（n = 本批消息数，闸门/指标未启用时 no-op）
	OnWriteBatch(n int)
	// DeferBroadcast 广播延迟重投（延迟队列，拒绝不等于丢弃）
	DeferBroadcast(ctx context.Context, msg *models.HubMessage, retry func(context.Context, *models.HubMessage))
	// GetBroadcastShaper 广播出向整形器（未启用时返回 nil）
	GetBroadcastShaper() *overload.Shaper
	// GetEphemeralCoalescer 高频消息合并器（未启用时返回 nil）
	GetEphemeralCoalescer() *overload.Coalescer
	// GetMessageAggregator 消息聚合器（未启用时返回 nil）
	GetMessageAggregator() *overload.MessageAggregator
	// GetOverloadMetrics 过载漏斗指标（编排层恒注入）
	GetOverloadMetrics() *overload.OverloadMetrics
	// GetMessageStatusUpdater 消息状态批量更新器（未注入时返回 nil）
	GetMessageStatusUpdater() *batcher.MessageStatusUpdater
	// GetAdmissionLevel 当前过载水位（闸门未启用时返回 LevelNormal）
	GetAdmissionLevel() overload.OverloadLevel
}

Host 消息域宿主端口

消息收发链路（发送/广播/分发/ACK 超时/离线转存）运行时依赖的 全部外部能力，由 hub 编排层实现并注入。

type HubWorkerPool 

type HubWorkerPool struct {
	// MessagePool 消息发送池（sendToUser、broadcast 等）
	MessagePool *syncx.WorkerPool
	// CallbackPool 业务回调池（连接/断开回调、事件回调）
	CallbackPool *syncx.WorkerPool
	// RecordPool 数据库记录池（消息记录、统计同步）
	RecordPool *syncx.WorkerPool
	// DistributedPool 跨节点消息处理池
	DistributedPool *syncx.WorkerPool
	// contains filtered or unexported fields
}

HubWorkerPool Hub 工作池集合 按任务类型分池，避免低优先级任务饿死高优先级任务

func NewHubWorkerPool 

func NewHubWorkerPool(cfg *wscconfig.WorkerPoolConfig, log spi.Logger) *HubWorkerPool

NewHubWorkerPool 创建 Hub 工作池集合 syncx.NewWorkerPool 在构造时自动启动 worker goroutine 弹性策略：构造仅启动保底数量，积压按需扩容至上限，空闲超过 cfg.IdleTimeout 自动回收

func (*HubWorkerPool) Stats 

func (w *HubWorkerPool) Stats() WorkerPoolStats

Stats 获取各池队列长度

func (*HubWorkerPool) Stop 

func (w *HubWorkerPool) Stop()

Stop 关闭所有工作池并等待任务完成

func (*HubWorkerPool) SubmitCallback 

func (w *HubWorkerPool) SubmitCallback(ctx context.Context, task func())

SubmitCallback 提交业务回调任务（阻塞式）

func (*HubWorkerPool) SubmitDistributed 

func (w *HubWorkerPool) SubmitDistributed(ctx context.Context, task func())

SubmitDistributed 提交跨节点消息处理任务（阻塞式）

func (*HubWorkerPool) SubmitMessage 

func (w *HubWorkerPool) SubmitMessage(ctx context.Context, task func())

SubmitMessage 提交消息发送任务（阻塞式，队列满时等待）

func (*HubWorkerPool) SubmitRecord 

func (w *HubWorkerPool) SubmitRecord(ctx context.Context, task func())

SubmitRecord 提交数据库记录任务（阻塞式）

func (*HubWorkerPool) TrySubmitCallback 

func (w *HubWorkerPool) TrySubmitCallback(task func()) bool

TrySubmitCallback 非阻塞提交业务回调任务

func (*HubWorkerPool) TrySubmitDistributed 

func (w *HubWorkerPool) TrySubmitDistributed(task func()) bool

TrySubmitDistributed 非阻塞提交跨节点消息处理任务

func (*HubWorkerPool) TrySubmitMessage 

func (w *HubWorkerPool) TrySubmitMessage(task func()) bool

TrySubmitMessage 非阻塞提交消息发送任务 队列满时返回 false，调用方可降级处理（如丢弃、记录失败）

func (*HubWorkerPool) TrySubmitRecord 

func (w *HubWorkerPool) TrySubmitRecord(task func()) bool

TrySubmitRecord 非阻塞提交数据库记录任务 队列满时丢弃（记录任务可丢失，不影响核心业务）

type HybridOfflineMessageConfig 

type HybridOfflineMessageConfig struct {
	RedisClient redis.UniversalClient // Redis 客户端（必需）
	DB          *gorm.DB              // MySQL 数据库（必需）
	KeyPrefix   string                // Redis key 前缀，默认 "wsc:offline:"
	QueueTTL    time.Duration         // Redis 队列过期时间，默认 7 天
	Logger      spi.Logger            // 日志器（可选）
}

HybridOfflineMessageConfig 混合存储配置

type HybridOfflineMessageHandler 

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

HybridOfflineMessageHandler 混合离线消息处理器 使用 Redis 队列存储短期离线消息（性能优先，快速推送） 使用 RDBMS offline_messages 表持久化（数据安全，防止 Redis 数据丢失） 注意：Redis 和 MySQL 必须同时初始化，双保险存储

func NewHybridOfflineMessageHandler 

func NewHybridOfflineMessageHandler(queue spi.MessageQueue, store spi.OfflineStore, config *wscconfig.OfflineMessage, log spi.Logger) *HybridOfflineMessageHandler

NewHybridOfflineMessageHandler 创建混合离线消息处理器

两个后端实现均由适配器注入 —— core 不构造任何具体存储实现， 依赖方向恒为 adapter → core。

参数:

  • queue: Redis 队列实现（必需，由 adapter/redis 注入）
  • store: RDBMS 持久化实现（必需，由 adapter/gorm 注入）
  • config: 离线消息配置对象
  • log: 日志记录器（nil 时使用默认）

func (*HybridOfflineMessageHandler) ClearOfflineMessages 

func (h *HybridOfflineMessageHandler) ClearOfflineMessages(ctx context.Context, userID string, groupIDs []string) error

ClearOfflineMessages 清空用户的离线消息 groupIDs: 用户在该 (appID, namespace) 下的全部 group（含 "" 表示 P2P 队列），逐组清 Redis + 一次清 MySQL

func (*HybridOfflineMessageHandler) DeleteOfflineMessages 

func (h *HybridOfflineMessageHandler) DeleteOfflineMessages(ctx context.Context, userID string, messageIDs []string) error

DeleteOfflineMessages 删除已推送的离线消息（appID/namespace 从 ctx 提取）

func (*HybridOfflineMessageHandler) DrainOfflineQueue 

func (h *HybridOfflineMessageHandler) DrainOfflineQueue(ctx context.Context, userID string, limit int) ([]*models.HubMessage, error)

DrainOfflineQueue 排空 Redis 队列（单组 FIFO，仅 Redis） 按 ctx 的单个 (appID, ns, group) 出队；limit<=0 表示一次取尽该队列

func (*HybridOfflineMessageHandler) GetOfflineMessageCount 

func (h *HybridOfflineMessageHandler) GetOfflineMessageCount(ctx context.Context, userID string) (int64, error)

GetOfflineMessageCount 获取离线消息数量（MySQL 跨组计数） MySQL 为双写超集（含 Redis 内 + Redis 已过期的全部待推送消息），故直接以 MySQL 计数为准

func (*HybridOfflineMessageHandler) GetOfflineMessages 

func (h *HybridOfflineMessageHandler) GetOfflineMessages(ctx context.Context, userID string, limit int, cursor string) ([]*models.HubMessage, string, error)

GetOfflineMessages 查询 MySQL 离线消息（跨组：命名空间内该用户全部 group） Redis 队列由 DrainOfflineQueue 单独排空；本方法只负责 MySQL 持久层的分页查询

参数:

  • userID: 用户ID
  • limit: >0 最多返回指定数量；<=0 最多 1 万条
  • cursor: 上次返回的最后一条 message_id，空串从头开始

返回 nextCursor 为空表示无更多数据

func (*HybridOfflineMessageHandler) StoreOfflineMessage 

func (h *HybridOfflineMessageHandler) StoreOfflineMessage(ctx context.Context, userID string, msg *models.HubMessage) error

StoreOfflineMessage 存储离线消息

多端登录场景说明： 当用户有多个设备（如ABC三个设备）时： - 如果ABC都离线：存储离线消息，任一设备上线时推送 - 如果AB在线C离线：消息已发送到AB，**不存储**离线消息，C上线后通过历史记录接口同步

核心原则： - **只有用户所有设备都离线时**，才存储离线消息并主动推送 - 有任何设备在线，消息已通过WebSocket实时送达，其他设备通过拉取历史记录获取 - 离线消息存储是基于用户维度的，用于在用户完全离线期间保证消息不丢失

去重机制： - 通过message_id保证消息唯一性（数据库unique索引） - 如果同一条消息重复存储，数据库层面会报错，但不影响功能

性能优化：Redis 和 MySQL 双写并行化

  • 原实现：顺序执行 storeToRedis + storeToDatabase，延迟 = T(Redis) + T(MySQL)
  • 现实现：并行执行，延迟 = max(T(Redis), T(MySQL))
  • msg 为只读（两个 goroutine 仅读取字段，无并发修改）
  • queueRepo（Redis 客户端）和 dbRepo（GORM 连接池）均为并发安全

func (*HybridOfflineMessageHandler) UpdatePushStatus 

func (h *HybridOfflineMessageHandler) UpdatePushStatus(ctx context.Context, messageIDs []string, pushErr error) error

UpdatePushStatus 更新离线消息推送状态 pushErr为nil表示推送成功,非nil表示推送失败

type Manager 

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

Manager 消息域管理器

组件注入采用「未注入即降级」语义：

  • ackManager 为 nil 时 ACK 快路径跳过
  • ackTimeoutTimer 为 nil 时跨节点 ACK 超时调度 no-op（兜底扫描仍在）
  • offlineHandler 为 nil 时投递兜底仅计数不转存
  • idGenerator 为 nil 时需调用方保证 msg.ID 非空

func NewManager 

func NewManager(host Host) *Manager

NewManager 创建消息域管理器

func (*Manager) BroadcastAfterDelay 

func (m *Manager) BroadcastAfterDelay(ctx context.Context, msg *models.HubMessage, delay time.Duration)

BroadcastAfterDelay 延迟广播消息（全局广播，走 Deliver）

func (*Manager) BroadcastByUserType 

func (m *Manager) BroadcastByUserType(ctx context.Context, userType models.UserType, msg *models.HubMessage) int

BroadcastByUserType 发送消息给特定用户类型的所有客户端（经 attrIdx 定点，自适应回退全量）

func (*Manager) BroadcastExclude 

func (m *Manager) BroadcastExclude(ctx context.Context, msg *models.HubMessage, excludeUserIDs []string) int

BroadcastExclude 广播消息给所有客户端，但排除指定用户

func (*Manager) BroadcastFrameToUserIDs 

func (m *Manager) BroadcastFrameToUserIDs(ctx context.Context, userIDs []string, msg *models.HubMessage, frame []byte) int

BroadcastFrameToUserIDs 预编码帧直发扇出（聚合 drain 专用零二次编码入口）

frame 为已编码的 WS 帧字节（如聚合器产出的裸 JSON 数组帧），逐成员 TrySend 直发 消除常规路径的二次 Marshal；msg 仍承担三件事：

  • 信封匹配键（AppID/Namespace/GroupIDs，per-device 过滤）
  • SSE 对象通道（TrySendSSE，SSE 无预编码通道）
  • TrySend 失败时的成员级分级兜底（routeDeliveryFallback）

frame 为空时回退一次 Marshal（与常规路径同语义，防御调用方未编码场景） 准入/整形/分片并行/顺序保证与 BroadcastToUserIDs 完全一致

func (*Manager) BroadcastPriority 

func (m *Manager) BroadcastPriority(ctx context.Context, msg *models.HubMessage, priority models.Priority)

BroadcastPriority 根据优先级广播消息（全局广播，走 Deliver）

func (*Manager) BroadcastToClientType 

func (m *Manager) BroadcastToClientType(ctx context.Context, clientType models.ClientType, msg *models.HubMessage) int

BroadcastToClientType 发送消息给特定客户端类型（经 attrIdx 定点，自适应回退全量）

func (*Manager) BroadcastToDepartment 

func (m *Manager) BroadcastToDepartment(ctx context.Context, department models.Department, msg *models.HubMessage) int

BroadcastToDepartment 发送消息给特定部门的所有用户（经 attrIdx 定点，自适应回退全量）

func (*Manager) BroadcastToFiltered 

func (m *Manager) BroadcastToFiltered(ctx context.Context, condition func(*models.Client) bool, msg *models.HubMessage) int

BroadcastToFiltered 预序列化消息并直接发送给符合条件的客户端 消除逐客户端 Clone/序列化/入队/DB 记录开销：

  • 消息只 json.Marshal 1 次（原方案每客户端 1 次）
  • 不走 SendToUserWithRetry（原方案每客户端 Clone×2 + 在线检查 + 入队 + DB 记录）
  • 零拷贝遍历（原方案 GetClientsCopy + FilterSlice 双重拷贝）

削峰填谷接线（分级准入 + 出向整形）：

  • Admit 拒绝（VerdictDelay/Offline）→ 延迟队列（填谷重投），队列满走分级兜底
  • shaper 拒绝（洪峰整形）→ 延迟队列；必达级跳过整形（与 Admit 矩阵的"必达恒放行"一致）

func (*Manager) BroadcastToNamespace 

func (m *Manager) BroadcastToNamespace(ctx context.Context, namespace string, msg *models.HubMessage) int

BroadcastToNamespace 命名空间广播统一入口（本地投递 deliverToNamespace 与 跨节点消费侧 hub.handleDistributedBroadcast 共用，保证两端路径一致）

经 nsIdx 反向索引定点反查该命名空间的本地在线用户后 BroadcastToUserIDs 扇出， 替代 BroadcastToFiltered 对全部连接的 O(N) 全量扫描：

  • 多租户 pod 上命名空间广播命中率通常极低（如 10w 连接中目标 ns 仅 10 个用户），索引将 O(全部连接) 收敛为 O(该 ns 本地在线用户)
  • 空命中直接返回（索引与注册表在同一 AddClient/RemoveClient 内事务性维护，无该 ns 用户时全量扫描同样 0 投递，跳过序列化与准入开销）

自适应回退：命中用户数 ≥ 总连接数的 nsBroadcastFullScanPercent% 时退回全量扫描 （索引路径逐 userID 查锁的单价更高，大命中率下 64 shard 并行批量遍历反而更快）

msg 信封在本入口统一同步（ContextWithRoute）：appID 归一化 + namespace 显式写入 确保 BroadcastToUserIDs 的 per-device 信封过滤（ClientMatchesEnvelope）与目标 ns 一致

func (*Manager) BroadcastToRole 

func (m *Manager) BroadcastToRole(ctx context.Context, role models.UserRole, msg *models.HubMessage) int

BroadcastToRole 发送消息给特定角色的所有用户（经 attrIdx 定点，自适应回退全量）

func (*Manager) BroadcastToSSEClients 

func (m *Manager) BroadcastToSSEClients(msg *models.HubMessage)

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

func (*Manager) BroadcastToUserIDs 

func (m *Manager) BroadcastToUserIDs(ctx context.Context, userIDs []string, msg *models.HubMessage) int

BroadcastToUserIDs 预序列化消息并直接发送给指定用户ID列表的在线客户端 O(m) 复杂度（m=用户数），按成员ID反查 shardedRegistry，仅锁定相关 shard 相比 BroadcastToFiltered 的 O(n)（n=总连接数），群组广播场景大幅减少遍历与锁范围 适用于已知目标用户ID列表的场景（群组广播、多群组广播）

削峰填谷接线：同 BroadcastToFiltered（分级准入 + 出向整形 + 延迟队列路由）

func (*Manager) CleanupExpiredAcks 

func (m *Manager) CleanupExpiredAcks() int

CleanupExpiredAcks 清理过期的 ACK 待确认消息（未注入 ACK 管理器时 no-op 返回 0）

func (*Manager) DeleteMessageRecord 

func (m *Manager) DeleteMessageRecord(ctx context.Context, id uint) error

DeleteMessageRecord 删除消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • id: 记录ID

func (*Manager) DeleteMessageRecordByMessageID 

func (m *Manager) DeleteMessageRecordByMessageID(ctx context.Context, messageID string) error

DeleteMessageRecordByMessageID 根据消息ID删除消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • messageID: 消息ID

func (*Manager) Deliver 

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

Deliver 统一消息投递入口 — 路由全由 ctx + msg 决定，一套逻辑打通所有场景

路由元数据来源：ctx（appID/namespace/groupIDs）+ msg（Receiver/RequireAck） 调用方通过 routing.NewRoute().WithAppID(appID).WithNamespace(ns).WithGroupIDs(gids).Inject(ctx) 注入路由后调一次 Deliver。

决策树（按优先级）：

  1. msg.Receiver != "" → P2P（SendToUserWithRetry，在线投递 + 离线存储 + 重试）
  2. len(groupIDs) > 0: - msg.RequireAck == true → 群组可靠投递（per-member SendToUserWithRetry + 离线存储） - msg.RequireAck == false → 群组广播（fire-and-forget，BroadcastToUserIDs + 跨节点）
  3. namespace != "" → 命名空间广播（BroadcastToFiltered + 跨节点 ns 广播）
  4. namespace == "" → 全局广播（handleBroadcast + clusterBatcher）

群组投递语义：ns 为来源租户记号（必传校验，81407），成员定位按 (appID, groupIDs) 跨 ns 聚合（业务侧按租户建组、同 gid 多实例，投递信封 ns 置空=通配，跨 ns 成员实时送达）

excludeSender 仅群组场景生效（P2P/广播场景 msg.Sender 不参与过滤）

返回非 nil *models.DeliverResult，错误收集到 result.Errors；按 result.Mode + 计数字段判断结果

func (*Manager) EndUserReplay 

func (m *Manager) EndUserReplay(userID string)

EndUserReplay 结束回放并按序补投暂存的实时投递（registry 任务闭包 defer 调用； 放弃回放权的触发方不得调用，否则会误开 inflight 回放的门闩）

func (*Manager) GetClientsByClientType 

func (m *Manager) GetClientsByClientType(clientType models.ClientType) []*models.Client

GetClientsByClientType 按客户端类型获取客户端（委托 FilterClients 零拷贝）

func (*Manager) GetClientsByDepartment 

func (m *Manager) GetClientsByDepartment(department models.Department) []*models.Client

GetClientsByDepartment 获取特定部门的所有客户端（委托 FilterClients 零拷贝）

func (*Manager) GetClientsByRole 

func (m *Manager) GetClientsByRole(role models.UserRole) []*models.Client

GetClientsByRole 获取特定角色的所有客户端（委托 FilterClients 零拷贝）

func (*Manager) GetClientsByUserType 

func (m *Manager) GetClientsByUserType(userType models.UserType) []*models.Client

GetClientsByUserType 获取特定用户类型的所有客户端（委托 FilterClients 零拷贝）

func (*Manager) GetClientsByVIPLevel 

func (m *Manager) GetClientsByVIPLevel(minVIPLevel models.VIPLevel) []*models.Client

GetClientsByVIPLevel 获取特定VIP等级及以上的客户端（委托 FilterClients 零拷贝）

func (*Manager) HandleBroadcastMessage 

func (m *Manager) HandleBroadcastMessage(ctx context.Context, msg *models.HubMessage)

HandleBroadcastMessage 处理广播消息

削峰填谷接线（与 broadcastToFiltered 同一套基座）：

  • 准入闸门（分级水位）：拒绝时延迟队列重投（填谷）/ 转离线，拒绝不等于丢弃
  • 出向整形（GCRA）：洪峰平滑扇出，被拒时进延迟队列；必达级跳过
  • 重投目标 doBroadcastMessage 不重复准入（防死循环，与 broadcastToFilteredNow 同语义）

func (*Manager) HoldUserDelivery 

func (m *Manager) HoldUserDelivery(userID string, deliver func())

HoldUserDelivery 用户级投递过闸（回放中暂存、开闸后立即执行）

func (*Manager) InvokeErrorCallback 

func (m *Manager) InvokeErrorCallback(ctx context.Context, err error, severity models.ErrorSeverity) error

InvokeErrorCallback 触发错误处理回调 此方法用于统一处理各种错误

func (*Manager) InvokeMessageReceivedCallback 

func (m *Manager) InvokeMessageReceivedCallback(ctx context.Context, client *models.Client, msg *models.HubMessage) error

InvokeMessageReceivedCallback 触发消息接收回调

func (*Manager) NotifyObservers 

func (m *Manager) NotifyObservers(ctx context.Context, msg *models.HubMessage)

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

func (*Manager) NotifyObserversDirect 

func (m *Manager) NotifyObserversDirect(msg *models.HubMessage, namespace string, groupIDs []string)

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

func (*Manager) OfferAggregatable 

func (m *Manager) OfferAggregatable(ctx context.Context, msg *models.HubMessage, excludeSender bool) bool

OfferAggregatable 可聚合档消息入窗尝试（群组广播链路统一接线点）

返回 true = 已入本地聚合窗口：调用方跳过即时本地扇出（drain 窗口批量扇出，各节点对本地成员各自聚合——源节点 + 跨节点入站流共用本入口） 返回 false = 不聚合或降级直通（调用方走原路径，性能降级不丢消息）：

  • 聚合器未启用 / 消息非可聚合档 / 多群组 / 排除发送者（窗口扇出无法复现逐成员排除语义）
  • 窗口满 / 群组数满（Offer 降级直通）
  • 序列化失败（编码前置一次，失败直通单发）

编码前置：入窗即完成本消息的序列化（drain 拼装为纯字节拼接，零重编码）

func (*Manager) PushOfflineMessages 

func (m *Manager) PushOfflineMessages(ctx context.Context, client *models.Client) (pushedIDs, failedIDs []string)

PushOfflineMessages 用户上线离线消息回放（两阶段全量补发）

全量补发策略（双写存储下的两阶段推送；全部状态在注入的共享存储， Deployment 滚动更新 / Pod 重新调度下跨 Pod 可见，Pod 本地零依赖）：

  1. 按组 drain Redis 队列（FIFO，短期高性能消息优先投递）：枚举用户在该 namespace 下加入的全部 group + P2P（空 group）队列，drain 出的消息推送 成功后按 message_id 删 MySQL（双写去重，避免阶段 2 重复推送）
  2. 跨组分页查 MySQL 剩余消息（Redis 已过期 / drain 异常残留），推送后删除

降级：groupRepo 不可用时只 drain P2P 队列 + 跨组查 MySQL（MySQL 跨组覆盖 所有 group 的消息，不丢消息，仅 Redis 短期队列残留待自然过期）

未注入离线处理器时返回空切片（nil-safe 降级）；推送失败的离线源消息不再 重新转存（send.go 离线源防循环），仅更新推送状态保留待下次上线重试。 返回成功/失败推送的 messageID 列表（编排层据此触发应用回调）

func (*Manager) QueryMessageRecord 

func (m *Manager) QueryMessageRecord(ctx context.Context, messageID, receiver string) (*models.MessageSendRecord, error)

QueryMessageRecord 根据消息ID+接收者查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • messageID: 消息ID
  • receiver: 接收者ID（广播类记录传空字符串）

返回:

  • *models.MessageSendRecord: 消息记录
  • error: 错误信息

func (*Manager) QueryMessageRecordsByClientIP 

func (m *Manager) QueryMessageRecordsByClientIP(ctx context.Context, clientIP string, limit int) ([]*models.MessageSendRecord, error)

QueryMessageRecordsByClientIP 根据客户端IP查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • clientIP: 客户端IP地址
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) QueryMessageRecordsByNodeIP 

func (m *Manager) QueryMessageRecordsByNodeIP(ctx context.Context, nodeIP string, limit int) ([]*models.MessageSendRecord, error)

QueryMessageRecordsByNodeIP 根据节点IP查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • nodeIP: 服务器节点IP
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) QueryMessageRecordsByReceiver 

func (m *Manager) QueryMessageRecordsByReceiver(ctx context.Context, receiver string, limit int) ([]*models.MessageSendRecord, error)

QueryMessageRecordsByReceiver 根据接收者查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • receiver: 接收者ID
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) QueryMessageRecordsBySender 

func (m *Manager) QueryMessageRecordsBySender(ctx context.Context, sender string, limit int) ([]*models.MessageSendRecord, error)

QueryMessageRecordsBySender 根据发送者查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • sender: 发送者ID
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) QueryMessageRecordsByStatus 

func (m *Manager) QueryMessageRecordsByStatus(ctx context.Context, status models.MessageSendStatus, limit int) ([]*models.MessageSendRecord, error)

QueryMessageRecordsByStatus 根据状态查询消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • status: 消息状态
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) QueryRetryableMessageRecords 

func (m *Manager) QueryRetryableMessageRecords(ctx context.Context, limit int) ([]*models.MessageSendRecord, error)

QueryRetryableMessageRecords 查询可重试的消息记录 参数:

  • ctx: 上下文（用于超时控制和取消）
  • limit: 返回结果数量限制（0 表示不限制）

func (*Manager) ScanNodeAckTimeouts 

func (m *Manager) ScanNodeAckTimeouts()

ScanNodeAckTimeouts 跨节点投递 ACK 超时兜底扫描（见 node_ack_timeout.go）

func (*Manager) ScheduleAckTimeouts 

func (m *Manager) ScheduleAckTimeouts(keys []models.MessageRecordKey)

ScheduleAckTimeouts 批量注册跨节点 ACK 超时任务（outbox flush 成功后回调， 仅传入状态仍为 sending 的记录键；注册延后 = outbox 攒批间隔，秒级超时语义无损）

func (*Manager) SendConditional 

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

SendConditional 根据条件发送消息给符合条件的客户端 复用 BroadcastToFiltered：预序列化一次 + 直接 TrySend，避免逐客户端 Clone/序列化/入队/DB 记录

func (*Manager) SendPriority 

func (m *Manager) SendPriority(ctx context.Context, userID string, msg *models.HubMessage, priority models.Priority)

SendPriority 根据优先级发送消息

func (*Manager) SendToAllClientsInMap 

func (m *Manager) SendToAllClientsInMap(ctx context.Context, clientMap map[string]*models.Client, msg *models.HubMessage)

SendToAllClientsInMap 发送消息到映射中的所有客户端 预序列化一次消息，避免对每个客户端重复 json.Marshal trace 恢复：ctx 无 trace 时从消息信封恢复（msg.ContextFrom），保证投递日志 与消息原始链路同一 trace_id（此前的 host.Context() 是 trace 断点）

func (*Manager) SendToClient 

func (m *Manager) SendToClient(ctx context.Context, client *models.Client, msg *models.HubMessage)

SendToClient 发送消息到客户端（内部序列化）

func (*Manager) SendToClientSerialized 

func (m *Manager) SendToClientSerialized(ctx context.Context, client *models.Client, msg *models.HubMessage, preSerialized []byte) bool

SendToClientSerialized 发送消息到客户端（支持预序列化数据） preSerialized 为预序列化的 []byte，为 nil 时内部序列化 SSE 客户端忽略 preSerialized，直接发送 msg 对象 返回是否成功投递到客户端通道（跨节点 PubSub 路径据此统计投递成败）

分级投递接线：

  • 高频级（Ephemeral）→ Coalescer latest-wins 合并（50ms 周期投递最新值）
  • Admit 拒绝（Delay/Offline，P2P 场景仅极端水位触发）→ 转离线补发 + 上层重试兜底
  • 实时路径零改动（TrySend 成功路径）

失败兜底按送达分级路由（与广播路径 TrySendWithFallback 同一语义）：

  • 高频级：丢弃计数（latest-wins 旧值过期是正确行为，转离线会让用户上线后收到一堆过期的弹幕/typing 状态消息）
  • 普通级：转离线补发（拒绝≠丢弃）

func (*Manager) SendToClientsWithRetry 

func (m *Manager) SendToClientsWithRetry(ctx context.Context, clients []*models.Client, msg *models.HubMessage, maxRetries int) map[string]*models.SendResult

SendToClientsWithRetry 发送消息给多个客户端（带重试） 使用预分配 slice + 索引定位写入，消除 mutex（每个 goroutine 写不同索引，无竞争）

func (*Manager) SendToGroupMembers 

func (m *Manager) SendToGroupMembers(ctx context.Context, memberIDs []string, msg *models.HubMessage, excludeSender bool) *models.BroadcastResult

SendToGroupMembers 向会话成员批量发送消息（兼容旧版本接口） 参数:

  • ctx: 上下文
  • memberIDs: 成员ID列表
  • msg: 要发送的消息
  • excludeSender: 是否排除发送者本身

返回:

  • models.BroadcastResult: 广播结果，包含成功、失败、离线统计

示例: 

向会话成员广播，排除发送者自己
result := hub.SendToGroupMembers(ctx, memberIDs, msg, true)
简单批量发送（不排除发送者）
result := hub.SendToGroupMembers(ctx, userIDs, msg, false)

func (*Manager) SendToMultipleUsers 

func (m *Manager) SendToMultipleUsers(ctx context.Context, userIDs []string, msg *models.HubMessage) map[string]error

SendToMultipleUsers 并发发送消息给多个用户 使用 runFanout 有界并发投递 + 预分配 slice + 索引写入，消除 mutex 竞争

func (*Manager) SendToUserViaSSE 

func (m *Manager) SendToUserViaSSE(userID string, msg *models.HubMessage) bool

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

func (*Manager) SendToUserWithRetry 

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

SendToUserWithRetry 带重试机制的发送消息给指定用户 公开 P2P 入口：先 EnsureRouteDefaults 归一化 appID/namespace（P2P 严格契约，空补默认值） 内部路径（sendToUserWithRetry/sendToUser）不再归一化——群组可靠投递的 per-member 扇出走同一管道，信封 ns="" 为跨 ns 通配语义（见 Deliver 群组分派器），归一化会破坏通配

func (*Manager) SendWithCallback 

func (m *Manager) SendWithCallback(ctx context.Context, userID string, msg *models.HubMessage,
	onSuccess func(*models.SendResult), onError func(error))

SendWithCallback 发送消息并在完成时执行回调

func (*Manager) StartClientPumps 

func (m *Manager) StartClientPumps(client *models.Client)

StartClientPumps 启动客户端读写泵并安装协议级 PING 处理器（编排层注册路径调用） 在注册临界区后调用：此后 socket 有了唯一读者（读泵）与唯一写者（写泵）， 泵启动前到达的消息由注册入口初始化的 SendChan 缓冲吸收，启动后排空，无丢失。 SSE 连接短路返回：SSE 无 gorilla 连接语义，写由 SSE 专用通道驱动（见 transport/sse.go）

func (*Manager) Stop 

func (m *Manager) Stop()

Stop 停止域内后台组件（工作池等由编排层持有的组件不在此列）

func (*Manager) StoreOfflineOnDeliveryFailure 

func (m *Manager) StoreOfflineOnDeliveryFailure(msg *models.HubMessage, deliveryErr error)

StoreOfflineOnDeliveryFailure 在在线投递失败时异步转存离线消息

触发条件（全部满足才转存）：

  • offlineMessageHandler 可用
  • msg.Receiver 非空（P2P 消息，广播场景不转存）
  • msg.Source != offline（离线推送本身失败不重新存，避免无限循环）

状态流转：

  • 转存成功 → message_record 状态从 Failed 覆盖为 UserOffline（消息已安全暂存，等用户上线推送）
  • 转存失败 → 保持调用方已标记的 Failed 状态（消息确实丢了）

注意：多设备场景下若部分设备投递成功部分失败，失败设备仍会触发转存 这不会导致重复推送：用户上线 drain 时 pushAndDeleteOffline 成功后按 message_id 删 MySQL，客户端也可按 message_id 去重

func (*Manager) SwapBroadcastFallbackCount 

func (m *Manager) SwapBroadcastFallbackCount() int64

SwapBroadcastFallbackCount 取出并清零广播兜底转存计数

func (*Manager) SwapBroadcastSentCount 

func (m *Manager) SwapBroadcastSentCount() int64

SwapBroadcastSentCount 取出并清零广播发送计数

func (*Manager) SwapMessageSentCount 

func (m *Manager) SwapMessageSentCount() int64

SwapMessageSentCount 取出并清零点对点消息发送计数

func (*Manager) TryBeginUserReplay 

func (m *Manager) TryBeginUserReplay(userID string) bool

TryBeginUserReplay 尝试标记用户进入首连离线回放，返回是否获得回放权 registry 首连分支调用：速断速连窗口同一用户可叠加多次首连触发（上线→回落→再上线） 并发回放会重复 drain 同一队列（离线消息重复投递）、且首个回放的 end 会提前开闸（乱序窗口重现）； inflight 回放刚 drain 过队列，重复触发放弃回放权是安全的（期间新转离线的消息由下次上线兜底）

func (*Manager) TryDeliverToClient 

func (m *Manager) TryDeliverToClient(ctx context.Context, client *models.Client, msg *models.HubMessage, data []byte) bool

TryDeliverToClient 返回实际准入或入队结果，离线兜底由汇总投递结果的调用方负责

func (*Manager) TrySendWithFallback 

func (m *Manager) TrySendWithFallback(client *models.Client, data []byte, msg *models.HubMessage, guarantee models.DeliveryGuarantee) bool

TrySendWithFallback 带分级兜底的非阻塞投递（广播扇出内部循环用）

性能契约：成功路径 = TrySend 一次（零新增指令）；失败路径才走兜底路由 guarantee 由调用方在扇出循环外预解析一次传入，复用成功路径送达埋点与失败兜底路由， 消除逐客户端重复 ResolveGuarantee 返回：是否实时送达（false = 已转离线/按高频语义处理，消息不会静默丢失）

func (*Manager) Wait 

func (m *Manager) Wait()

Wait 等待全部读写泵 goroutine 退出（编排层关闭时调用）

func (*Manager) WithAckManager 

func (m *Manager) WithAckManager(am *AckManager) *Manager

WithAckManager 注入 ACK 管理器

func (*Manager) WithAckTimeoutTimer 

func (m *Manager) WithAckTimeoutTimer(t *syncx.HashedWheelTimer) *Manager

WithAckTimeoutTimer 注入跨节点 ACK 超时时间轮

func (*Manager) WithErrorCallback 

func (m *Manager) WithErrorCallback(cb ErrorCallback) *Manager

WithErrorCallback 注入错误处理回调

func (*Manager) WithIDGenerator 

func (m *Manager) WithIDGenerator(g models.IDGenerator) *Manager

WithIDGenerator 注入消息 ID 生成器

func (*Manager) WithMessageReceivedCallback 

func (m *Manager) WithMessageReceivedCallback(cb MessageReceivedCallback) *Manager

WithMessageReceivedCallback 注入消息接收回调

func (*Manager) WithMessageSendCallback 

func (m *Manager) WithMessageSendCallback(cb MessageSendCallback) *Manager

WithMessageSendCallback 注入消息发送完成回调

func (*Manager) WithOfflineHandler 

func (m *Manager) WithOfflineHandler(h spi.OfflineQueue) *Manager

WithOfflineHandler 注入离线消息处理器（spi.OfflineQueue 契约；混合实现见 HybridOfflineMessageHandler）

func (*Manager) WithReplayGate 

func (m *Manager) WithReplayGate(g *replayGate) *Manager

WithReplayGate 注入首连回放门闩（未注入时投递直通，行为与旧版一致）

func (*Manager) WithWorkerPool 

func (m *Manager) WithWorkerPool(wp *HubWorkerPool) *Manager

WithWorkerPool 注入工作池集合

type MessageReceivedCallback 

type MessageReceivedCallback func(ctx context.Context, client *models.Client, msg *models.HubMessage) error

MessageReceivedCallback 消息接收回调（客户端上行消息交给业务层处理）

type MessageSendCallback 

type MessageSendCallback func(msg *models.HubMessage, result *models.SendResult)

MessageSendCallback 消息发送完成回调

type OfflineDeps 

type OfflineDeps struct {
	// Queue Redis 队列实现（如 redisadapter.NewOfflineQueue(...)）
	Queue spi.MessageQueue
	// Store RDBMS 持久化实现（如 gormadapter.NewOfflineStoreFor(...)）
	Store spi.OfflineStore
	// Config 离线消息配置
	Config *wscconfig.OfflineMessage
	// Logger 日志器（nil 使用默认）
	Logger spi.Logger
}

OfflineDeps 混合离线消息处理器的两半依赖

该处理器同时需要 Redis 队列与 RDBMS 持久化，但两者分属不同适配器， 单个适配器的 hook 都拿不到完整依赖 —— 故由本函数在两侧 hook 跑完后统一组装。

type PendingMessage 

type PendingMessage struct {
	Message   *models.HubMessage // 原始消息
	AckChan   chan *AckMessage   // ACK确认通道
	Timestamp time.Time          // 发送时间
	Timeout   time.Duration      // 超时时间
	Retry     int                // 重试次数
	MaxRetry  int                // 最大重试次数
	// contains filtered or unexported fields
}

PendingMessage 待确认消息

func (*PendingMessage) WaitForAck 

func (pm *PendingMessage) WaitForAck() (*AckMessage, error)

WaitForAck 等待ACK确认

func (*PendingMessage) WaitForAckWithRetry 

func (pm *PendingMessage) WaitForAckWithRetry(retryFunc func() error) (*AckMessage, error)

WaitForAckWithRetry 等待ACK确认并支持重试

type WorkerPoolStats 

type WorkerPoolStats struct {
	MessageQueueLen     int
	CallbackQueueLen    int
	RecordQueueLen      int
	DistributedQueueLen int
}

WorkerPoolStats 工作池统计信息

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.