overload

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

README

overload（过载域）

AIMD 削峰填谷 + 按用户批量投递——流量进来时的分级裁决，与出口的批量发送

这个域干什么

overload 在消息洪峰与后端容量之间做缓冲：入口用准入闸门和速率整形削峰， 出口把同用户的多条消息合并批量投递batcher.攒批落库已拆分至独立的 batcher 域batcher.

核心概念

入口半边：削峰（REFACTORING_DESIGN §五.7）
消息入口 → AdmissionGate(AIMD) → BroadcastShaper(GCRA)
              │                        │
              ▼                        ▼
         分级裁决              DelayQueue(填谷)
        (VIP/普通)            Coalescer(Latest-Wins 合并)
概念 说明
AdmissionGate AIMD 准入：水位线 + 快降慢升（连续 N 周期确认才恢复），防抖动
Shaper GCRA 虚调度令牌桶：CAS 单点更新 tat，无锁零分配
Coalescer Latest-wins 合并：同 key 覆盖旧值，16 分片防内存膨胀
BroadcastDelayQueue 填谷延迟队列：超载消息延迟放行而非丢弃
FallbackAction 降级动作：VIP 直通 / 普通延迟 / 低优丢弃

队列优先级容量：system 10% / critical 15% / high 25% / normal 35% / low 15%

出口半边：批量投递
概念 说明
BatchSender 按用户聚合批量发送：同用户多条消息合并为一次投递
BatchSendResult / UserResult 批量发送结果与单用户结果
BatchSendFailureCallback 失败回调：批量发送失败时逐用户回报

怎么用

package main

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

// 入口：准入 + 整形
gate := overload.NewAdmissionGate(high, low, evalInterval)
if gate.Admit(priority) != overload.AdmitPass {
    // 走 FallbackAction：延迟 / 降级 / 丢弃
}

shaper := overload.NewShaper(constants.DefaultBroadcastShaperRate)
ok := shaper.Allow()

端口与依赖

  • 消费者端口：UserMessageSender（hub 实现，BatchSender 的单用户投递能力）
  • 指标经 OverloadMetrics 暴露，由编排层恒注入消息域

性能与陷阱

  • P10：GCRA 多 worker 必须 CAS 单点更新 tat，否则整形失效
  • AIMD 参数：降载立即生效，恢复要连续 N 周期确认——两者不对称是防抖的关键
  • Coalescer 只适合可丢弃的 ephemeral 消息（Latest-Wins 覆盖旧值）

后端绑定

  • 无直接存储绑定：本域纯内存组件，落库由 batcher 域与 spi 契约承担

Documentation

Overview

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-11 19:30:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-11 20:30:00 * @FilePath: \go-wsc\overload\admission.go * @Description: 准入闸门 —— 分级驱动的削峰填谷核心 * * 机制（AIMD + 分级决策矩阵）： * - backlog = delivered - written（在途量，Little's law：backlog/吞吐 ≈ 排队延迟） * - 每 evalInterval 评估一次：backlog > 高水位 → level++（快速升载，过载保护优先） * 连续 cooldownLevels 个周期 backlog < 低水位 → level--（缓慢降载，防抖动） * - Admit 按分级×水位矩阵给出三态裁决：放行（实时）/ 延迟（入延迟队列）/ 离线（转离线补发） * * 设计原则：拒绝≠丢弃——过载时按消息分级路由兜底（延迟/离线），全链路无静默丢弃 * * 性能红线： * - Admit 零分配（读 atomic level + 查表返回枚举，无 map/无 heap） * - 埋点仅 atomic add（OnWriteBatch 每批 1 次、OnDelivered 每投递 1 次） * - 未启用（nil gate）时接线层直接跳过，零开销 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-10-08 01:56:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-10 19:56:00 * @FilePath: \go-wsc\overload\aggregator.go * @Description: 消息聚合器 —— per-group 窗口攒批 + 数组零重编码单帧推送（N 条合 1 条） * * 与 Coalescer（latest-wins，丢旧保新，适配行情/输入状态）互补： * 聚合语义是全保留合并——窗口内消息按序拼装为 JSON 数组单帧推送， * 每连接帧率 = 窗口频率（聚合后推送数 = 在线连接数，内核 pps 友好） * * 聚合维度与投递路由统一：聚合 key 就是群组广播的 groupID（不引入房间 * 等平行概念），聚合产出直接回流群组扇出链路；弹幕/跑马灯/滚动通知等 * 一切"容忍百毫秒延迟、全保留合并"的群组流式消息均可接入 * * 编码前置：Offer 接收的是调用方已序列化的消息字节，数组拼装为纯字节 * 拼接（[p1,p2,...]），零重编码——1000w msg/s 输入下编码次数 = 消息数 * * 降级语义（性能降级不丢消息）：单群组窗口满 / 群组数超限时 Offer 返回 * false，调用方直发单条；空群组在 Flush 换出空批时惰性删除 * * 并发模型：16 分片 × per-group 独立锁；Offer 持 shard 锁嵌套 group 锁完成入窗（消除取引用与空窗口删除之间的竞态窗口，防消息写入已删除的孤儿群组）； * Flush 同序（shard→group 单向，无死锁），锁内仅双缓冲 swap，数组装配锁外执行（大窗口 memcpy 不阻塞入窗） * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-09 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-09 00:50:00 * @FilePath: \go-wsc\overload\coalescer.go * @Description: 高频消息合并器 —— latest-wins 语义（同 key 只保留最新） * * 高频流式消息（正在输入/已读/状态信号）语义上只需最新值到达： * - Offer(key, msg)：同 key 直接覆盖旧消息（旧值天然过期，覆盖≠丢弃） * - Drain()：批量取出当前所有最新值（投递入口周期性/水位驱动调用） * * 并发模型：单写者覆盖（Offer 持分片锁覆盖 map 槽位），Drain 换出整个 map * （swap-and-rebuild，O(容量) 一次拷贝，投递侧遍历新表无锁） * 分片锁降低 Offer 热点竞争（同 ShardedMap 模式，shard = FNV(按字段流式) % N） * * 容量保护：条目数超限（高频 key 滥用/攻击）时 Offer 返回 false， * 上层按高频级语义丢弃计数（latest-wins 的尽头是容量上限，防内存膨胀） * * 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-10 21:00:00 * @FilePath: \go-wsc\overload\delay_queue.go * @Description: 广播延迟队列 —— 填谷填充器（削峰的"谷"侧） * * 整形拒绝（shaper.Allow() == false）的广播消息进入此队列延迟投递： * - 有界容量（BroadcastDelayQueueCapacity），满则走分级兜底路由（转离线/高频计数），不丢弃必达 * - 单 drain goroutine：每轮取 1 条 + 按 shaper 速率节拍投递 * （洪峰时排队削峰；水位回落后 AIMD 提速 → drain 节拍加快 → 填谷加速） * * 性能：入队 1 次 chan send（非阻塞，满即 fallback）；drain 单协程无锁竞争 * * 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-12 21:57:19 * @FilePath: \go-wsc\overload\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-09 21:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-10 21:00:00 * @FilePath: \go-wsc\overload\overload_metrics.go * @Description: 过载观测指标 —— 送达漏斗与投递路由分支（全 atomic 零锁计数） * * 送达漏斗（per Guarantee 分级）： * admitted（准入放行） → realtime（实时送达）/ offline（转离线补发） * ephemeral-merged（latest-wins 合并）/ ephemeral-dropped（高频语义丢弃） * unrecoverable（无兜底可用时的最终丢失——仅离线 handler 未配置的部署问题） * * 投递路由分支（Deliver 决策树五模式 + 跨节点通道）： * deliver_modes：P2P / 群组可靠 / 群组广播 / 命名空间 / 全局（消息级日志降 Debug 后的观测补位） * cluster_grpc：跨节点 gRPC 直连投递（消息数与累计节点数，均扇出 = nodes/messages） * cluster_pubsub_fallbacks：PubSub 兜底发布次数（gRPC 未覆盖时的降级频率） * cluster_dispatch_syncs：集群分发队列满时可靠档同步直发降级次数（分布式池反压观测） * * 守恒不变量（送达保证的量化验收，测试断言）： * 必达/普通级：realtime + offline + unrecoverable == admitted * 高频级：merged + dropped == 高频 admitted * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-11 22:16:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-09-11 23:59:00 * @FilePath: \go-wsc\overload\tokenbucket.go * @Description: GCRA 令牌桶 —— 广播出向整形（削峰填谷的"谷"填充器） * * GCRA（Generic Cell Rate Algorithm）虚调度实现： * - 仅 1 个 atomic.Int64（theoretical arrival time 纳秒值），无桶无队列 * - Allow() 单次 CAS 循环：允许则推进 tat，拒绝则不动 * - burst 容量通过容忍的提前量（burst offset）实现 * * 与经典令牌桶的区别：无锁（单 atomic CAS）、无定时补充协程、零分配； * SetRate 支持运行时热调整（AIMD 联动：过载降速/恢复提速） * * 用途：广播扇出前每条消息 1 令牌（非每客户端），洪峰时平滑扇出速率； * 被拒消息进有界延迟队列（填谷时 AIMD 提速加速 drain），不丢弃 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

Index

Constants

View Source 
const BroadcastDelayQueueCapacity = 1024

BroadcastDelayQueueCapacity 延迟队列容量（有界；超过走分级兜底路由）

Variables

View Source 
var ErrDeliveryFallback = errors.New("delivery fallback: send channel full")

errDeliveryFallback 兜底转存的触发错误（语义标记，供状态更新/日志使用）

Functions

This section is empty.

Types

type AdmissionGate 

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

AdmissionGate 准入闸门

零值/默认构造即可工作（NewAdmissionGate 填充默认水位），未 Start 时 level 恒为 L0（Admit 全放行），Start 后开始水位评估

func NewAdmissionGate 

func NewAdmissionGate(highWatermark, lowWatermark int64, evalInterval time.Duration) *AdmissionGate

NewAdmissionGate 创建准入闸门（默认水位开箱即用，参数零值时用 constants 默认值）

func (*AdmissionGate) Admit 

func (g *AdmissionGate) Admit(msg *models.HubMessage, isBroadcast bool) AdmitVerdict

Admit 消息级准入裁决（零分配热路径：atomic load + 查表）

决策矩阵（分级 × 水位 → 放行/延迟/离线）： 

水位             必达级          普通级           高频级
L0 正常          放行            放行             放行（合并后）
L1 延迟普通      放行            延迟             放行（合并本身削峰）
L2 延迟广播      放行            延迟/离线        放行（合并后）
L3 仅关键        放行            离线             放行（合并后）
L4 只读          放行*           离线             离线**

* L4 必达级仍放行：控制通道独立（CtrlCh cap 16），KickOut/断链语义不受数据面过载影响
** L4 高频级：转离线无意义（状态类语义只需最新），由 Coalescer latest-wins 兜底

func (*AdmissionGate) Backlog 

func (g *AdmissionGate) Backlog() int64

Backlog 当前在途量（绑定的实际积压计数为唯一真实源） 未绑定 counter 时恒为 0：水位评估恒低于低水位、Admit 恒放行（安全退化，仅 Stats 观测有效） 不再用 delivered-written 累计差兜底：断连放弃帧"已计投递但永不写出"，累计差会 单调漂移使水位虚高卡死在 L4——独立使用闸门必须 SetPendingCounter 绑定实际积压计数

func (*AdmissionGate) CooldownLevels 

func (g *AdmissionGate) CooldownLevels() int

CooldownLevels 返回降载冷却层级数（AIMD 迟滞参数）

func (*AdmissionGate) EvalInterval 

func (g *AdmissionGate) EvalInterval() time.Duration

OnWriteBatch 写泵批埋点（每批 1 次 atomic add，batch 条消息共用） Hub 的 nil-safe 转发方法见 hub.go 的 OnWriteBatch EvalInterval 返回评估周期（供上层对齐扫描节拍）

func (*AdmissionGate) ForceLevel 

func (g *AdmissionGate) ForceLevel(level OverloadLevel)

ForceLevel 强制设置水位（测试/运维手动降载用）

func (*AdmissionGate) HighWatermark 

func (g *AdmissionGate) HighWatermark() int64

HighWatermark 返回高水位阈值（积压超过即升载）

func (*AdmissionGate) Level 

func (g *AdmissionGate) Level() OverloadLevel

Level 当前水位（无锁读，观测/日志用）

func (*AdmissionGate) LowWatermark 

func (g *AdmissionGate) LowWatermark() int64

LowWatermark 返回低水位阈值（积压低于即降载）

func (*AdmissionGate) OnDelivered 

func (g *AdmissionGate) OnDelivered()

OnDelivered 投递埋点（每条消息成功入队后调用，1 次 atomic add）

func (*AdmissionGate) OnWriteBatch 

func (g *AdmissionGate) OnWriteBatch(n int)

func (*AdmissionGate) SetPendingCounter 

func (g *AdmissionGate) SetPendingCounter(counter *atomic.Int64)

SetPendingCounter 在闸门启动前绑定 Hub 的实际积压计数（启用水位评估的前置条件）

func (*AdmissionGate) Start 

func (g *AdmissionGate) Start()

Start 启动水位评估循环（幂等：重复调用无效果）

func (*AdmissionGate) Stats 

func (g *AdmissionGate) Stats() AdmissionStats

Stats 获取状态快照

func (*AdmissionGate) Stop 

func (g *AdmissionGate) Stop()

Stop 停止评估循环（Hub 关闭时调用）

type AdmissionStats 

type AdmissionStats struct {
	Level     OverloadLevel
	Backlog   int64
	Written   int64
	Delivered int64
}

Stats 闸门状态快照（观测用）

type AdmitVerdict 

type AdmitVerdict int

AdmitVerdict 准入裁决（零分配枚举，三态） 

const (
	// VerdictAdmit 放行（实时投递）
	VerdictAdmit AdmitVerdict = iota
	// VerdictDelay 延迟（入有界延迟队列，水位回落后加速 drain——填谷）
	VerdictDelay
	// VerdictOffline 转离线（离线补发链路，上线时推送——仍是送达）
	VerdictOffline
)

type AggregatedBatch 

type AggregatedBatch struct {
	AppID   string
	GroupID string
	Frame   []byte // JSON 数组体 "[p1,p2,...]"（元素为已序列化消息字节，零重编码）
}

AggregatedBatch 一批聚合产出（单群组一个窗口的全部消息）

type BatchSendFailureCallback 

type BatchSendFailureCallback func(userID string, msg *models.HubMessage, err error)

BatchSendFailureCallback 批量发送单条消息失败时的回调

type BatchSendResult 

type BatchSendResult struct {
	TotalUsers    int                    // 总用户数
	TotalMessages int                    // 总消息数
	SuccessCount  int32                  // 成功发送的消息数
	FailureCount  int32                  // 失败的消息数
	UserResults   map[string]*UserResult // 每个用户的发送结果
}

BatchSendResult 批量发送结果

type BatchSender 

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

BatchSender 批量消息发送器

func New 

func New(user UserMessageSender, failureCallback BatchSendFailureCallback, ctx context.Context) *BatchSender

New 创建批量发送器

failureCallback 由调用方（Hub）在构造时注入 —— 它来自 Hub 的未导出字段， 放在此处注入可避免为它新增一个公开访问器。

func (*BatchSender) AddMessage 

func (bs *BatchSender) AddMessage(userID string, msg *models.HubMessage) *BatchSender

AddMessage 为用户添加一条消息（支持链式调用）

func (*BatchSender) AddMessages 

func (bs *BatchSender) AddMessages(userID string, msgs ...*models.HubMessage) *BatchSender

AddMessages 为用户添加多条消息（支持链式调用）

func (*BatchSender) AddUserMessages 

func (bs *BatchSender) AddUserMessages(userMessages map[string][]*models.HubMessage) *BatchSender

AddUserMessages 批量添加用户消息映射（支持链式调用）

func (*BatchSender) Clear 

func (bs *BatchSender) Clear() *BatchSender

Clear 清空所有待发送消息

func (*BatchSender) Count 

func (bs *BatchSender) Count() (users int, messages int)

Count 获取当前待发送的消息统计

func (*BatchSender) Execute 

func (bs *BatchSender) Execute() *BatchSendResult

Execute 执行批量发送

func (*BatchSender) ExecuteAsync 

func (bs *BatchSender) ExecuteAsync(callback func(*BatchSendResult))

ExecuteAsync 异步执行批量发送（不等待结果）

type BroadcastDelayQueue 

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

broadcastDelayQueue 有界延迟队列

func NewBroadcastDelayQueue 

func NewBroadcastDelayQueue(shaper func() ShaperInterval) *BroadcastDelayQueue

NewBroadcastDelayQueue 创建延迟队列（drain 循环由调用方启动）

func (*BroadcastDelayQueue) Backlog 

func (q *BroadcastDelayQueue) Backlog() int64

Backlog 队列当前积压（观测用）

func (*BroadcastDelayQueue) DrainLoop 

func (q *BroadcastDelayQueue) DrainLoop(ctx context.Context)

drainLoop 投递循环（每轮 1 条 + 按 shaper 速率节拍等待）

节拍：每投递 1 条 sleep 一个令牌间隔（interval = 1s / rate）； AIMD 恢复时 SetRate 提升 → interval 缩短 → 填谷加速（闭环联动）

func (*BroadcastDelayQueue) Offer 

func (q *BroadcastDelayQueue) Offer(ctx context.Context, msg *models.HubMessage, retry func(ctx context.Context, msg *models.HubMessage)) bool

offer 入队一条延迟投递（非阻塞；返回 false = 队列满，调用方走分级兜底路由）

func (*BroadcastDelayQueue) Stats 

func (q *BroadcastDelayQueue) Stats() map[string]int64

Stats 快照（观测用）

type Coalescer 

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

Coalescer 高频消息合并器（latest-wins）

func NewCoalescer 

func NewCoalescer(capacity int) *Coalescer

NewCoalescer 创建合并器（capacity <= 0 时用默认容量）

func (*Coalescer) Capacity 

func (c *Coalescer) Capacity() int

Capacity 返回合并器总容量上限

func (*Coalescer) Drain 

func (c *Coalescer) Drain() []*models.HubMessage

Drain 取出当前所有最新消息（swap-and-rebuild）

返回合并后的消息列表（可空）；调用方投递后由下一轮 Offer 重新累积 高频消息的投递语义：本轮 Drain 的即当前最新快照

func (*Coalescer) MergedCount 

func (c *Coalescer) MergedCount() int64

MergedCount 累计合并（覆盖）次数——latest-wins 削峰效果的直接量化

func (*Coalescer) Offer 

func (c *Coalescer) Offer(key EphemeralKey, msg *models.HubMessage) (accepted, merged bool)

Offer 提交一条高频消息（latest-wins：同 key 覆盖旧消息）

返回 (accepted, merged)：

  • accepted=false：容量超限（防滥用）或参数非法，上层按高频语义丢弃计数
  • merged=true：本次覆盖了同 key 旧消息（latest-wins 削峰，供漏斗 merged 埋点）

func (*Coalescer) Size 

func (c *Coalescer) Size() int64

Size 当前累积条目数（观测用）

type EphemeralKey 

type EphemeralKey struct {
	UserID      string
	MessageType models.MessageType
}

EphemeralKey 高频合并 key（用户 × 消息类型：同用户同类型只保最新值）

结构化 key：map 查找与分片哈希均按字段进行，Offer 热路径零字符串分配 （拼接式 key 在百万级高频扇出下是持续的 GC 分配热点）

type FallbackAction 

type FallbackAction int
const (
	// FallbackNone 无需兜底（不应出现：调用方误用）
	FallbackNone FallbackAction = iota
	// FallbackOffline 转离线补发（必达/普通级）
	FallbackOffline
	// FallbackEphemeralDrop 高频级丢弃（latest-wins 语义正确行为）
	FallbackEphemeralDrop
)

type MessageAggregator 

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

MessageAggregator 消息聚合器（per-group 窗口攒批，N 条合 1 条数组帧） 零值可用（构造经 NewMessageAggregator 兜底默认容量）； drain 节拍与产出投递归编排层（hub 的周期 ticker 调 Flush）

func NewMessageAggregator 

func NewMessageAggregator(maxPerGroup, maxGroups int) *MessageAggregator

NewMessageAggregator 创建消息聚合器 maxPerGroup <= 0 时用 constants.DefaultAggregatorBatchSize； maxGroups <= 0 时用 constants.DefaultAggregatorMaxGroups

func (*MessageAggregator) Flush 

func (a *MessageAggregator) Flush() []AggregatedBatch

Flush 换出全部窗口攒批并拼装为 JSON 数组帧（drain ticker 周期调用）

返回本窗口的全部聚合批（可空）；空窗口的群组惰性删除（无人发言的 群组在换出后回收，群组数随活跃群组自然收敛）； 拼装为纯字节拼接，元素零重编码

锁内仅做双缓冲 swap（O(1) 指针交换），数组装配全程锁外执行： 换出走 spare 交换（ taken 缓冲底层数组在归还前仅 Flush 持有，并发 Offer 写入的是 swap 后的新缓冲），大窗口 MB 级 memcpy 不再阻塞入窗路径 单写者契约：Flush 仅由 drain ticker 单 goroutine 周期调用，swap 无竞争

func (*MessageAggregator) Offer 

func (a *MessageAggregator) Offer(appID, groupID string, payload []byte) bool

Offer 提交一条已序列化的消息进入窗口攒批

返回 false = 降级直通（调用方单发该条，性能降级不丢消息）：

  • 单群组窗口满（maxPerGroup）：洪峰下窗口来不及 Flush
  • 群组数超限（maxGroups）：群组风暴保护
  • 参数非法（空 appID/groupID/载荷）

锁结构：shard 锁内嵌套 group 锁完成入窗——若先释放 shard 锁再拿 group 锁，Flush 可能在此间隙删除空窗口群组，append 会写入已从 map 摘除的孤儿对象导致消息永久丢失（并发守恒测试可复现）

func (*MessageAggregator) Stats 

func (a *MessageAggregator) Stats() map[string]int64

Stats 快照（观测用）

type OverloadLevel 

type OverloadLevel int32

OverloadLevel 过载水位等级（atomic.Int32 存储，热路径无锁读）

升级快（每周期最多 +1 但立即生效）、降级慢（连续 N 周期低于低水位才 -1）， 与 TCP 拥塞控制的 AIMD 语义一致：快速响应过载、缓慢恢复防抖动 

const (
	// LevelNormal L0：正常，全部实时投递
	LevelNormal OverloadLevel = iota
	// LevelDelayStandard L1：延迟普通级（普通消息入延迟队列，必达/高频照常）
	LevelDelayStandard
	// LevelDelayBroadcast L2：延迟广播（广播走慢速整形，普通级延迟/转离线）
	LevelDelayBroadcast
	// LevelCriticalOnly L3：仅关键（普通级全部转离线，必达走控制通道，高频只发最新）
	LevelCriticalOnly
	// LevelReadOnly L4：只读（仅控制通道可投递，普通级转离线，高频丢弃计数）
	LevelReadOnly
)

func (OverloadLevel) String 

func (l OverloadLevel) String() string

String 水位等级的日志友好表示

type OverloadMetrics 

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

OverloadMetrics 过载与送达漏斗指标（零值可用，全 atomic 无锁）

func (*OverloadMetrics) DeliveryFunnelConsistent 

func (m *OverloadMetrics) DeliveryFunnelConsistent() (bool, map[string]int64)

DeliveryFunnelConsistent 送达漏斗守恒校验（测试/自检断言）

必达/普通级守恒：realtime + offline_fallback + unrecoverable == admitted （高频级走 latest-wins 覆盖语义，覆盖发生在 admitted 计数之后，精确守恒无意义，不校验） 返回 false 时返回的 map 描述不守恒的分级与差值（运维告警信号）

func (*OverloadMetrics) MetricsSnapshot 

func (m *OverloadMetrics) MetricsSnapshot() map[string]int64

MetricsSnapshot 标量计数快照（测试/自检断言用；分桶计数见 OverloadStats）

func (*OverloadMetrics) OverloadStats 

func (m *OverloadMetrics) OverloadStats() map[string]any

OverloadStats 送达漏斗与过载指标快照（map 形态，并入现有 stats 查询体系）

func (*OverloadMetrics) RecordAdmissionVerdict 

func (m *OverloadMetrics) RecordAdmissionVerdict(v AdmitVerdict)

RecordAdmissionVerdict 准入裁决埋点（延迟/离线三态分流）

func (*OverloadMetrics) RecordAdmitted 

func (m *OverloadMetrics) RecordAdmitted(g models.DeliveryGuarantee)

RecordAdmitted 准入埋点（每成员 1 次，分片写入）

func (*OverloadMetrics) RecordClusterDispatchSync 

func (m *OverloadMetrics) RecordClusterDispatchSync()

RecordClusterDispatchSync 集群分发队列满时可靠档同步直发降级埋点

func (*OverloadMetrics) RecordClusterGRPC 

func (m *OverloadMetrics) RecordClusterGRPC(nodes int)

RecordClusterGRPC 跨节点 gRPC 直连投递埋点（nodes = 本次成功直连的节点数） 一次 routeToCluster 成功直连 ≥1 节点即计 1 条跨节点消息；均扇出 = nodes/messages

func (*OverloadMetrics) RecordClusterPubSubFallback 

func (m *OverloadMetrics) RecordClusterPubSubFallback()

RecordClusterPubSubFallback PubSub 兜底发布埋点（gRPC 未覆盖节点时的成功发布口径）

func (*OverloadMetrics) RecordDeliverMode 

func (m *OverloadMetrics) RecordDeliverMode(mode models.DeliveryMode)

RecordDeliverMode Deliver 路由分支埋点（入口决策树分类后调用，消息数维度）

func (*OverloadMetrics) RecordEphemeralDrop 

func (m *OverloadMetrics) RecordEphemeralDrop()

RecordEphemeralDrop 高频语义丢弃埋点

func (*OverloadMetrics) RecordEphemeralMerged 

func (m *OverloadMetrics) RecordEphemeralMerged()

RecordEphemeralMerged 高频 latest-wins 合并埋点（Offer 覆盖同 key 旧消息时调用）

func (*OverloadMetrics) RecordOfflineFallback 

func (m *OverloadMetrics) RecordOfflineFallback(g models.DeliveryGuarantee)

RecordOfflineFallback 转离线补发埋点

func (*OverloadMetrics) RecordRealtime 

func (m *OverloadMetrics) RecordRealtime(g models.DeliveryGuarantee)

RecordRealtime 实时送达埋点

func (*OverloadMetrics) RecordShaperDenied 

func (m *OverloadMetrics) RecordShaperDenied()

RecordShaperDenied 整形拒绝埋点（进延迟队列）

func (*OverloadMetrics) RecordSlowEvict 

func (m *OverloadMetrics) RecordSlowEvict()

RecordSlowEvict 慢消费者驱逐埋点

func (*OverloadMetrics) RecordUnrecoverable 

func (m *OverloadMetrics) RecordUnrecoverable(g models.DeliveryGuarantee)

RecordUnrecoverable 无兜底丢失埋点（正常为 0；非 0 = 离线 handler 未配置的部署缺陷）

func (*OverloadMetrics) RecordWriteBatch 

func (m *OverloadMetrics) RecordWriteBatch(n int)

RecordWriteBatch 写泵批量埋点（n = 本批消息数）

type Shaper 

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

Shaper GCRA 令牌桶（出向整形器）

零值不可直接使用（rate 未设置），NewShaper 构造

func NewShaper 

func NewShaper(ratePerSec int64) *Shaper

NewShaper 创建整形器（ratePerSec <= 0 时用默认广播整形速率）

func (*Shaper) Allow 

func (s *Shaper) Allow() bool

Allow 是否放行一个令牌（GCRA 单次 CAS，无锁零分配）

GCRA 判定：now + burstTime >= tat 则放行并推进 tat = max(tat, now) + interval 突发场景 tat 领先 now（排队虚拟时间）；稳态 tat 与 now 同步推进

func (*Shaper) Interval 

func (s *Shaper) Interval() int64

interval 每令牌间隔（纳秒）；rate<=0 返回 0 表示不整形

func (*Shaper) Rate 

func (s *Shaper) Rate() int64

Rate 当前速率（观测用）

func (*Shaper) SetRate 

func (s *Shaper) SetRate(ratePerSec int64)

SetRate 热调整速率（AIMD 联动：过载时降速、水位恢复时提速） min 1（0 表示关闭整形，与关闭语义冲突的由 SetShaping 控制）

type ShaperInterval 

type ShaperInterval interface {
	Interval() int64
}

ShaperInterval 整形器速率查询（interval 返回当前令牌间隔，纳秒）

type UserMessageSender 

type UserMessageSender interface {
	// SendToUserWithRetry 向指定用户发送消息（带重试）
	SendToUserWithRetry(ctx context.Context, toUserID string, msg *models.HubMessage) *models.SendResult
}

UserMessageSender 用户消息发送能力（BatchSender 所需）

type UserResult 

type UserResult struct {
	UserID        string
	TotalMessages int
	SuccessCount  int
	FailureCount  int
	Errors        []error
}

UserResult 用户发送结果

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.