ratelimit

Documentation

Overview

Package ratelimit 提供通用限流器实现。

支持令牌桶和滑动窗口两种限流算法，以及按 key 隔离的限流器管理； Lockout 提供按 key 的防爆破锁（fail2ban 形态：失败计数达阈值 → 锁定拒绝态）。

用法:

// 令牌桶限流器
limiter := ratelimit.NewTokenBucket(100, 200) // 100 QPS, burst 200
if limiter.Allow() {
    // 处理请求
}

// 按 key 隔离的限流器（如按用户 ID）
keyed := ratelimit.NewKeyedLimiter(func() ratelimit.Limiter {
    return ratelimit.NewTokenBucket(10, 20)
}, 10*time.Minute)
if keyed.Allow("user:123") {
    // 处理请求
}

// 防爆破锁（登录/API Key 防爆破、匿名接口防枚举）
lo := ratelimit.NewLockout(ratelimit.LockoutConfig{
    Window: 10 * time.Minute, Threshold: 5, LockDuration: 15 * time.Minute,
})
if locked, remain := lo.CheckLocked("user:42"); locked {
    // 对外 429 + 剩余秒数文案（remain 即剩余锁定时长）
}
if locked, _ := lo.RecordFailure("user:42"); locked {
    // 本次失败触发锁定（或已在锁定中）
}
lo.Reset("user:42") // 登录成功清零

Index

• type KeyedLimiter
  • func NewKeyedLimiter(factory func() Limiter, ttl time.Duration) *KeyedLimiter
• type KeyedLimiterOf
  • func NewKeyedLimiterByFunc[K comparable](factory func(K) Limiter, opts ...KeyedOption[K]) *KeyedLimiterOf[K]
  • func NewKeyedLimiterOf[K comparable](factory func() Limiter, opts ...KeyedOption[K]) *KeyedLimiterOf[K]
  • func (kl *KeyedLimiterOf[K]) Allow(key K) bool
  • func (kl *KeyedLimiterOf[K]) AllowN(key K, n int) bool
  • func (kl *KeyedLimiterOf[K]) AllowWithRetryAfter(key K) (wait time.Duration, ok bool)
  • func (kl *KeyedLimiterOf[K]) Close()
  • func (kl *KeyedLimiterOf[K]) Create(key K, l Limiter) bool
  • func (kl *KeyedLimiterOf[K]) Delete(key K)
  • func (kl *KeyedLimiterOf[K]) Get(key K) Limiter
  • func (kl *KeyedLimiterOf[K]) Len() int
  • func (kl *KeyedLimiterOf[K]) OverCapHits() int64
  • func (kl *KeyedLimiterOf[K]) Replace(key K, l Limiter)
• type KeyedOption
  • func WithIdleTTL[K comparable](ttl time.Duration) KeyedOption[K]
  • func WithMaxEntries[K comparable](max int) KeyedOption[K]
  • func WithOverCapRetryAfter[K comparable](d time.Duration) KeyedOption[K]
• type Limiter
• type Lockout
  • func NewLockout(cfg LockoutConfig) *Lockout
• type LockoutConfig
• type LockoutOf
  • func NewLockoutOf[K comparable](cfg LockoutConfig) *LockoutOf[K]
  • func (l *LockoutOf[K]) CheckLocked(key K) (locked bool, lockRemain time.Duration)
  • func (l *LockoutOf[K]) Close()
  • func (l *LockoutOf[K]) Len() int
  • func (l *LockoutOf[K]) OverCapHits() int64
  • func (l *LockoutOf[K]) RecordFailure(key K) (locked bool, lockRemain time.Duration)
  • func (l *LockoutOf[K]) Reset(key K)
• type SlidingWindow
  • func NewSlidingWindow(window time.Duration, maxCount int) *SlidingWindow
  • func (sw *SlidingWindow) Allow() bool
  • func (sw *SlidingWindow) AllowN(n int) bool
  • func (sw *SlidingWindow) Count() int
  • func (sw *SlidingWindow) Reserve() time.Duration
  • func (sw *SlidingWindow) Reset()
• type TokenBucket
  • func NewTokenBucket(rps rate.Limit, burst int, opts ...TokenBucketOption) *TokenBucket
  • func (tb *TokenBucket) Allow() bool
  • func (tb *TokenBucket) AllowN(n int) bool
  • func (tb *TokenBucket) Burst() int
  • func (tb *TokenBucket) Rate() rate.Limit
  • func (tb *TokenBucket) Reserve() time.Duration
  • func (tb *TokenBucket) RetryAfter() (time.Duration, bool)
• type TokenBucketOption
  • func WithTokenClock(clock func() time.Time) TokenBucketOption

Types

type KeyedLimiter

type KeyedLimiter = KeyedLimiterOf[string]

KeyedLimiter 是 string key 的别名，兼容既有调用方。

func NewKeyedLimiter

func NewKeyedLimiter(factory func() Limiter, ttl time.Duration) *KeyedLimiter

NewKeyedLimiter 创建 string key 的限流器管理器（兼容旧 API）。

参数:

• factory: 创建新限流器的工厂函数
• ttl: 限流器空闲过期时间，超过此时间未使用的限流器会被清理

用法:

keyed := ratelimit.NewKeyedLimiter(func() ratelimit.Limiter {
    return ratelimit.NewTokenBucket(10, 20)
}, 10*time.Minute)

type KeyedLimiterOf

type KeyedLimiterOf[K comparable] struct {
	// contains filtered or unexported fields
}

KeyedLimiterOf 是按 key 隔离的限流器管理器（key 类型可定制）。 每个 key 维护独立的限流器实例，支持空闲回收、容量护栏、速率热更。

func NewKeyedLimiterByFunc

func NewKeyedLimiterByFunc[K comparable](factory func(K) Limiter, opts ...KeyedOption[K]) *KeyedLimiterOf[K]

NewKeyedLimiterByFunc 创建按 key 隔离、且**配额由 key 决定**的限流器管理器（#31）。

NewKeyedLimiterOf 的 factory 签名是 func() Limiter —— 不接受 key，因此一个管理器内 所有 key 只能共享同一份配额。业务上按接口维度分档位（同步类宽松 / 备份类严格 / 导入确认最严）是常见形态，此前只能把限流器实例的键从 (maxReq, window) 扩成 (action, maxReq, window)，即每个 action 一个管理器 —— 那样按 key 隔离、空闲回收、 容量护栏、Len()/OverCapHits() 可观测出口这些价值基本被绕过，整体限流状态也没有 单一治理点。本构造函数是它的正解：一个管理器，按 key 定配额。

用法:

quota := map[string]struct {
    maxReq  int
    window  time.Duration
}{
    "backup/import/confirm": {2, 900 * time.Second},
    "car/syncAllTripData":   {3, 300 * time.Second},
}
keyed := ratelimit.NewKeyedLimiterByFunc[string](func(k string) ratelimit.Limiter {
    q := quota[k]
    return ratelimit.NewSlidingWindow(q.window, q.maxReq)
}, ratelimit.WithIdleTTL[string](10*time.Minute))
ok := keyed.Allow("car/syncAllTripData")   // 该 key 用自己那档配额

⚠️ 未在配额表里登记的 key 会拿到 factory 返回的值（若 factory 写了兜底默认值， 就是静默按兜底档限流）。这是 API 形状使然，不是库的 bug——但它曾导致一次迁移中 未登记的接口被按默认档限流。**配额表缺 key 时应让它显式失败**（返回 nil 会让 Allow 返回 false，见 Get 的容量护栏说明），或在 Allow 处自行判空，避免静默降级。

func NewKeyedLimiterOf

func NewKeyedLimiterOf[K comparable](factory func() Limiter, opts ...KeyedOption[K]) *KeyedLimiterOf[K]

NewKeyedLimiterOf 创建按 key 隔离的限流器管理器（key 任意 comparable 类型）， 所有 key 共用同一份限流配额。

需要**按 key 给不同配额**（不同接口不同档位）请用 NewKeyedLimiterByFunc（#31）。

用法:

keyed := ratelimit.NewKeyedLimiterOf[uint](
    func() ratelimit.Limiter { return ratelimit.NewTokenBucket(10, 20) },
    ratelimit.WithIdleTTL[uint](10*time.Minute),
    ratelimit.WithMaxEntries[uint](10_000),
)

func (*KeyedLimiterOf[K]) Allow

func (kl *KeyedLimiterOf[K]) Allow(key K) bool

Allow 检查指定 key 是否允许请求。容量护栏拒绝未登记 key 时返回 false。

func (*KeyedLimiterOf[K]) AllowN

func (kl *KeyedLimiterOf[K]) AllowN(key K, n int) bool

AllowN 检查指定 key 是否允许 N 个请求。容量护栏拒绝未登记 key 时返回 false。

func (*KeyedLimiterOf[K]) AllowWithRetryAfter

func (kl *KeyedLimiterOf[K]) AllowWithRetryAfter(key K) (wait time.Duration, ok bool)

AllowWithRetryAfter 检查是否放行；拒绝时返回建议等待时长（不消耗令牌的探测）。

• ok=true：请求放行，wait=0
• ok=false：请求拒绝；wait 是下一个可用令牌的等待时长（Reserve→Delay→Cancel， 不消耗令牌），容量护栏拒绝时返回建议 Retry-After

探测语义：被拒请求不会自我消耗配额，避免放大限流（429 + Retry-After 场景）。

func (*KeyedLimiterOf[K]) Close

func (kl *KeyedLimiterOf[K]) Close()

Close 关闭限流器管理器，停止清理 goroutine。可安全重复调用。

func (*KeyedLimiterOf[K]) Create

func (kl *KeyedLimiterOf[K]) Create(key K, l Limiter) bool

Create 在容量护栏内原子登记新 key 的限流器（[#33](https://github.com/oekit/concurrency/issues/33)）： 表满先回收空闲条目（与 Allow 建桶路径同一 sweep-on-pressure），仍满返回 false 并计一次 OverCapHits。已登记的 key 返回 false 且**不覆盖**——速率热更请走 Replace（换桶不该被 护栏卡住）。l 为 nil 返回 false。

与 NewKeyedLimiterByFunc 互补：ByFunc 的 factory 在库内锁内被调用，消费方若在 factory 里读自持锁的速率状态会形成「库锁→消费方锁」与正常路径「消费方锁→库锁」的 锁序倒置；「每次调用传入当前配置速率」的消费方（Nacos/DB 热更即时生效）应外部构造 桶后用 Create 登记 / Replace 热更，配额回收与护栏都留在库内。

func (*KeyedLimiterOf[K]) Delete

func (kl *KeyedLimiterOf[K]) Delete(key K)

Delete 删除指定 key 的限流器。

func (*KeyedLimiterOf[K]) Get

func (kl *KeyedLimiterOf[K]) Get(key K) Limiter

Get 获取指定 key 的限流器。表满且 key 未登记时返回 nil。

func (*KeyedLimiterOf[K]) Len

func (kl *KeyedLimiterOf[K]) Len() int

Len 返回当前活跃的限流器数量。

func (*KeyedLimiterOf[K]) OverCapHits

func (kl *KeyedLimiterOf[K]) OverCapHits() int64

OverCapHits 返回容量护栏拒绝未登记 key 的累计次数（可观测性）。

func (*KeyedLimiterOf[K]) Replace

func (kl *KeyedLimiterOf[K]) Replace(key K, l Limiter)

Replace 将指定 key 的限流器替换为新实例（速率热更）。 换桶即新配额，桶内余量归零（调用方接受）。key 不存在时等价登记新桶。

type KeyedOption

type KeyedOption[K comparable] func(*keyedConfig[K])

KeyedOption 是 KeyedLimiterOf 的函数选项。

func WithIdleTTL

func WithIdleTTL[K comparable](ttl time.Duration) KeyedOption[K]

WithIdleTTL 设置空闲回收周期；表满时也先触发回收再判满。 <=0 使用默认 1 分钟。

小于 2ns 的值不会让后台清扫 panic——清扫节奏有 1s 下限（见 cleanupInterval）， 过期判定仍按本值走。

func WithMaxEntries

func WithMaxEntries[K comparable](max int) KeyedOption[K]

WithMaxEntries 设置容量护栏上限（未登记 key 在表满时被拒绝，已登记 key 不受影响）。 0 = 不限（默认）。

func WithOverCapRetryAfter

func WithOverCapRetryAfter[K comparable](d time.Duration) KeyedOption[K]

WithOverCapRetryAfter 设置容量护栏拒绝时 AllowWithRetryAfter 返回的建议等待。 <=0 使用默认 1 分钟。

type Limiter

type Limiter interface {
	// Allow 检查是否允许一个请求。
	Allow() bool

	// AllowN 检查是否允许 N 个请求。
	AllowN(n int) bool

	// Reserve 预留一个令牌，返回需要等待的时间。
	// 如果返回 0 表示立即可用。
	Reserve() time.Duration
}

Limiter 是通用限流器接口。

type Lockout

type Lockout = LockoutOf[string]

Lockout 是 string key 的别名，兼容惯例（对齐 KeyedLimiter）。

func NewLockout

func NewLockout(cfg LockoutConfig) *Lockout

NewLockout 创建 string key 的防爆破锁。 Window/Threshold/LockDuration 为必填语义参数，非法值直接 panic （配置错误宁可启动即暴露——锁静默失效等于没有防护）。

type LockoutConfig

type LockoutConfig struct {
	// Window 为失败计数窗口：窗口内的失败事件按 key 累计（滑动窗口，
	// 半开区间 (now-Window, now]，窗口外样本自然老化）。
	Window time.Duration
	// Threshold 为锁定阈值：窗口内第 N 次失败触发锁定。
	Threshold int
	// LockDuration 为锁定时长：触发后 key 进入拒绝态持续 L 秒，期间 CheckLocked
	// 返回剩余时长（供对外 429 文案）；到期自动解除。后续失败不延长锁定。
	LockDuration time.Duration
	// MaxEntries 为容量护栏上限（0 = 不限，对齐 KeyedLimiterOf.WithMaxEntries）。
	// 表满且 key 未登记时失败样本被丢弃（fail-open + OverCapHits 观测）——
	// 不因内部容量问题拒绝认证尝试（那会放大成对合法用户的拒绝服务）。
	MaxEntries int
	// IdleTTL 为空闲回收周期；<=0 用默认 1 分钟。锁定中的条目免于回收
	// （提前驱逐等于提前解锁）；窗口内仍有有效失败样本的条目同样免于回收
	// （#83：空闲回收线不得短于失败窗口，否则滑窗语义被截短，阈值可被
	// 「间隔 > IdleTTL」的分布式攻击绕过）；回收不改变锁定到期语义。
	// 小于 2ns 的值不会让后台清扫 panic——清扫节奏有 1s 下限（见 cleanupInterval），
	// 空闲判定仍按本值走。
	IdleTTL time.Duration
	// Now 为时钟注入（nil = time.Now；与 budget.WithClock 同款，
	// 假时钟测试完全确定）。清扫节奏仍走真实 ticker，到期/空闲判定一律用本时钟。
	Now func() time.Time
}

LockoutConfig 是 Lockout 的构造配置。

type LockoutOf

type LockoutOf[K comparable] struct {
	// contains filtered or unexported fields
}

LockoutOf 是按 key 隔离的防爆破锁（fail2ban 形态）：窗口内失败事件达阈值 → key 进入锁定拒绝态持续 LockDuration（期间即使凭证正确也应拒绝）→ 成功操作 Reset 清零。与家族原语的边界：KeyedLimiter 是速率型令牌桶（表达不了 「计数达标后进入持续拒绝态」）；breaker 是调用方视角的实例级熔断（无 key 隔离）； budget 是超限记账存档（无拒绝态与 Reset）。

语义细则：

• 失败样本滑动窗口计数，窗口外自然老化——锁定到期后若旧失败仍在窗口内， 首个新失败即重新锁定（重复违规者语义，对齐 fail2ban findtime 行为）；
• 锁定期间新失败只入窗口记账（封顶 Threshold 个），不延长当前锁定；
• 容量护栏 fail-open：表满且 key 未登记时丢弃失败样本并计入 OverCapHits；
• 锁定中的条目免于空闲回收（提前驱逐等于提前解锁）；窗口内仍有有效失败 样本的条目同样免于回收——空闲回收只清理「已无窗口内样本」的条目， 否则窗口语义会被 IdleTTL 截短（#83）。

典型用法（登录防爆破）：

lo := ratelimit.NewLockout(ratelimit.LockoutConfig{
    Window: 10 * time.Minute, Threshold: 5, LockDuration: 15 * time.Minute,
    MaxEntries: 10_000,
})
if locked, remain := lo.CheckLocked("user:42"); locked {
    return &LockedError{RemainingSeconds: int(remain.Seconds())} // 对外 429
}
if err := verifyCredential(); err != nil {
    if locked, _ := lo.RecordFailure("user:42"); locked { ... }
    return err
}
lo.Reset("user:42") // 登录成功

func NewLockoutOf

func NewLockoutOf[K comparable](cfg LockoutConfig) *LockoutOf[K]

NewLockoutOf 创建任意 comparable key 的防爆破锁（IP、用户名、API Key 等）。

func (*LockoutOf[K]) CheckLocked

func (l *LockoutOf[K]) CheckLocked(key K) (locked bool, lockRemain time.Duration)

CheckLocked 查询 key 是否处于锁定拒绝态：locked=true 时 lockRemain 为剩余 锁定时长（供对外 429 文案），未锁定或未登记返回 (false, 0)。查询刷新空闲计时。 返回序与 RecordFailure 统一为 (locked, lockRemain)——两个方法同序是刻意设计， 避免调用方在相邻调用间交换返回值顺序。

func (*LockoutOf[K]) Close

func (l *LockoutOf[K]) Close()

Close 停止后台清扫 goroutine。可安全重复调用（KeyedLimiterOf 同款幂等语义）。

func (*LockoutOf[K]) Len

func (l *LockoutOf[K]) Len() int

Len 返回当前活跃条目数（含锁定中）。

func (*LockoutOf[K]) OverCapHits

func (l *LockoutOf[K]) OverCapHits() int64

OverCapHits 返回容量护栏丢弃失败样本的累计次数（可观测性）。

func (*LockoutOf[K]) RecordFailure

func (l *LockoutOf[K]) RecordFailure(key K) (locked bool, lockRemain time.Duration)

RecordFailure 记一次失败事件。

• 触发锁定或 key 已处于锁定态：返回 (true, 剩余锁定时长)；
• 未触发：返回 (false, 0)；
• 容量护栏拒绝（表满且 key 未登记）：失败样本被丢弃，返回 (false, 0)， 可经 OverCapHits 观测。

锁定期间调用只入窗口记账，不延长当前锁定。

func (*LockoutOf[K]) Reset

func (l *LockoutOf[K]) Reset(key K)

Reset 清零 key 的失败计数并解除现存锁定（成功操作后调用；亦可用作管理员解封）。 key 未登记为 no-op。

type SlidingWindow

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

SlidingWindow 是滑动窗口限流器。 在指定时间窗口内限制请求次数。

func NewSlidingWindow

func NewSlidingWindow(window time.Duration, maxCount int) *SlidingWindow

NewSlidingWindow 创建滑动窗口限流器。

参数:

• window: 时间窗口大小
• maxCount: 窗口内最大请求数

用法:

limiter := ratelimit.NewSlidingWindow(time.Minute, 100) // 每分钟最多 100 次

func (*SlidingWindow) Allow

func (sw *SlidingWindow) Allow() bool

Allow 检查是否允许一个请求。

func (*SlidingWindow) AllowN

func (sw *SlidingWindow) AllowN(n int) bool

AllowN 检查是否允许 N 个请求。n<=0 视为 0 个请求，直接放行（不占槽位）。

func (*SlidingWindow) Count

func (sw *SlidingWindow) Count() int

Count 返回当前窗口内的请求数。

func (*SlidingWindow) Reserve

func (sw *SlidingWindow) Reserve() time.Duration

Reserve 预留一个请求，返回需要等待的时间。 返回 0 表示立即放行（槽位已计入窗口）；返回 d>0 表示需等待 d 后放行， 槽位已按 now+d 预占——与 x/time/rate 的 Reserve 语义一致：预留即消费， 并发调用者拿到的是互不相同的槽位而非同一个"虚拟许可"。

func (*SlidingWindow) Reset

func (sw *SlidingWindow) Reset()

Reset 重置限流器。

type TokenBucket

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

TokenBucket 是令牌桶限流器。 基于 golang.org/x/time/rate 实现，支持平滑限流；时间源可注入 WithTokenClock （缺省 time.Now），确定性测试注入假时钟即可无 sleep 驱动配额恢复与 Retry-After 探测（#24）。

func NewTokenBucket

func NewTokenBucket(rps rate.Limit, burst int, opts ...TokenBucketOption) *TokenBucket

NewTokenBucket 创建令牌桶限流器。

参数:

• rps: 每秒允许的请求数（Requests Per Second）
• burst: 突发容量（令牌桶最大容量）

用法:

limiter := ratelimit.NewTokenBucket(100, 200) // 100 QPS, burst 200

func (*TokenBucket) Allow

func (tb *TokenBucket) Allow() bool

Allow 检查是否允许一个请求。

func (*TokenBucket) AllowN

func (tb *TokenBucket) AllowN(n int) bool

AllowN 检查是否允许 N 个请求。

func (*TokenBucket) Burst

func (tb *TokenBucket) Burst() int

Burst 返回限流器的 burst 配置。

func (*TokenBucket) Rate

func (tb *TokenBucket) Rate() rate.Limit

Rate 返回限流器的 RPS 配置。

func (*TokenBucket) Reserve

func (tb *TokenBucket) Reserve() time.Duration

Reserve 预留一个令牌，返回需要等待的时间。

func (*TokenBucket) RetryAfter

func (tb *TokenBucket) RetryAfter() (time.Duration, bool)

RetryAfter 探测下一个可用令牌的等待时长，不消耗令牌（Reserve→Delay→Cancel， 归还按同一注入时钟结算）。ok=false 表示限流器本身不可用（如 burst=0 且 rps=0）。 被拒请求不会自我消耗配额——429 + Retry-After 场景的标准姿势。

type TokenBucketOption

type TokenBucketOption func(*TokenBucket)

TokenBucketOption 用于定制 TokenBucket（函数选项模式；按本仓命名硬规则为 XxxOption，禁止裸 Option）。

func WithTokenClock

func WithTokenClock(clock func() time.Time) TokenBucketOption

WithTokenClock 注入时钟（缺省 time.Now；nil 视为缺省）。放行判定、预留与 Retry-After 探测全部走该时钟——假时钟下「1 秒后应恢复配额」「Retry-After 精确回填」均可确定性地断言，无 sleep 无 flake。