先复盘：429 是怎么被打出来的
订单导出服务要给第三方物流接口逐个下单，对方文档写着 100 QPS。我们的代码长这样：
// 反例:5 个 worker 把 5000 个请求推出去,没有任何限流
var wg sync.WaitGroup
sem := make(chan struct{}, 5)
for _, o := range orders {
wg.Add(1)
go func(o Order) {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
callVendorAPI(o) // 单次耗时约 20ms
}(o)
}
wg.Wait()
算一下：5 个 worker、单次 20ms，稳态就是 250 QPS，超了对方限额的 2.5 倍。这还没完——收到 429 之后我们的重试逻辑 for i := 0; i < 3; i++（间隔 100ms）把实际发出量抬到 750 QPS，对方直接把我们丢进黑名单，整个导出任务全线卡死。
所以限流要解决的问题不是「保护自己」，而是别让自己的重试放大器把下游打穿。下面从实现往回讲。
令牌桶：按时间差补令牌，别起定时器
package ratelimit
import (
"sync"
"time"
)
// TokenBucket 令牌桶。
// rate —— 每秒补充的令牌数,可以是小数(0.5 表示每 2 秒一个)
// burst —— 桶容量,同时也是允许的最大瞬时突发量
type TokenBucket struct {
mu sync.Mutex
rate float64
burst float64
tokens float64
lastTime time.Time
}
func NewTokenBucket(rate float64, burst int) *TokenBucket {
if rate <= 0 || burst <= 0 {
panic("ratelimit: rate and burst must be > 0")
}
return &TokenBucket{
rate: rate,
burst: float64(burst),
tokens: float64(burst), // 冷启动装满,允许一次突发
lastTime: time.Now(),
}
}
// AllowN 尝试取走 n 个令牌,不阻塞。
func (b *TokenBucket) AllowN(n int) bool {
if n <= 0 {
return true
}
if float64(n) > b.burst {
// 单次要的比桶还多,永远不可能成功。直接拒,别让调用方空转重试。
return false
}
b.mu.Lock()
defer b.mu.Unlock()
now := time.Now() // 必须在锁内取,原因见下面「坑三」
elapsed := now.Sub(b.lastTime).Seconds()
if elapsed < 0 {
elapsed = 0
}
b.lastTime = now
// 核心:一次性把这段时间该补的令牌补上,不需要任何后台 goroutine。
// time.Now() 带单调时钟,Sub 的结果不受 NTP 校时影响。
b.tokens += elapsed * b.rate
if b.tokens > b.burst {
b.tokens = b.burst // 超出容量的部分直接丢弃,不能攒着
}
if b.tokens < float64(n) {
return false
}
b.tokens -= float64(n)
return true
}
func (b *TokenBucket) Allow() bool { return b.AllowN(1) }
这段代码的关键只有一句话：令牌不是被「推」进来的，是被「算」出来的。每次调用时用 now - lastTime 算出这段空窗期该补多少令牌，一次补齐。没有 goroutine，没有定时器，空闲时不消耗任何资源。
坑一：用后台 ticker 补令牌
// 反例:每个限流器配一个 1ms 的定时器补令牌
func NewBadBucket(rate, burst int) *BadBucket {
b := &BadBucket{burst: burst}
go func() {
t := time.NewTicker(time.Millisecond)
defer t.Stop()
for range t.C {
b.mu.Lock()
if b.tokens < b.burst {
b.tokens++
}
b.mu.Unlock()
}
}()
return b
}
单看一个限流器好像没什么。但如果按商户维度限流，服务里有 500 个商户就是 500 个 goroutine、每秒 50 万次定时器唤醒加锁——而你真正要限的请求可能每秒只有几千个。而且 1ms 的粒度意味着限流精度被锁死在 1000 QPS 以下，想限 5000 QPS 根本做不到。惰性计算版本是零后台开销，且精度只受 float64 限制。
坑二：用 int64 存令牌数
// 反例
b.tokens += int64(elapsed * b.rate)
rate = 0.5（每 2 秒一个令牌）、调用间隔 1.5 秒时，elapsed * rate = 0.75，转 int64 被截断成 0，令牌永远补不上，这个限流器会一直拒绝。低速限流在「某个接口每天只能调 100 次」这类场景很常见，必须用 float64 存令牌。
坑三：now 在锁外取
// 反例
now := time.Now()
b.mu.Lock()
defer b.mu.Unlock()
b.tokens += now.Sub(b.lastTime).Seconds() * b.rate
b.lastTime = now
高并发下几十个 goroutine 堵在锁上，每个人手里的 now 都比它真正拿到锁的时刻早。lastTime 被反复改成一个偏早的值，累积下来实际放行速率会明显低于设定值——你设了 100 QPS，跑出来可能只有 60。先把锁拿到手再取 now，误差就只有锁内那几十纳秒。
漏桶有两种，混着讲会出事
大部分「令牌桶 vs 漏桶」的对比文章其实在拿两种不同的东西比。漏桶至少有下面两个实现，行为差得很远。
实现 A：准入式漏桶
// LeakyBucket 请求能否进来只看水位,进来的请求立即放行。
type LeakyBucket struct {
mu sync.Mutex
rate float64
capacity float64
water float64
lastTime time.Time
}
func NewLeakyBucket(rate float64, capacity int) *LeakyBucket {
return &LeakyBucket{
rate: rate,
capacity: float64(capacity),
lastTime: time.Now(),
}
}
func (b *LeakyBucket) Allow() bool {
b.mu.Lock()
defer b.mu.Unlock()
now := time.Now()
elapsed := now.Sub(b.lastTime).Seconds()
if elapsed < 0 {
elapsed = 0
}
b.lastTime = now
// 这段时间漏掉的水,先从水位里扣掉
b.water -= elapsed * b.rate
if b.water < 0 {
b.water = 0
}
if b.water+1 > b.capacity {
return false
}
b.water++
return true
}
说句得罪人的话：这个实现和令牌桶在数学上几乎等价。令牌桶的数从 burst 往下减、空闲时往上补；漏桶的数从 0 往上加、空闲时往下漏。同一套公式换个方向看而已。空闲 10 秒后，两者都允许 burst（或 capacity）个请求一口气发出。
所以如果你看到某篇文章说「令牌桶允许突发、漏桶不允许」，它八成拿的是下面这个版本在比。
实现 B：整形式漏桶（这才是真正的区别）
var ErrBucketFull = errors.New("leaky bucket is full")
// ShapingLeakyBucket 请求排队,后台以固定间隔依次放行,输出端严格平滑。
type ShapingLeakyBucket struct {
queue chan chan struct{}
done chan struct{}
once sync.Once
}
func NewShapingLeakyBucket(ratePerSec float64, queueSize int) *ShapingLeakyBucket {
b := &ShapingLeakyBucket{
queue: make(chan chan struct{}, queueSize),
done: make(chan struct{}),
}
go b.drain(ratePerSec)
return b
}
// drain 每秒放行 ratePerSec 个请求。
// 注意:定时器间隔低于 1ms 时,调度器唤醒抖动会占很大比例,
// 实际放行速率会明显低于设定值。这个实现适合几百到几千 QPS 的量级。
func (b *ShapingLeakyBucket) drain(ratePerSec float64) {
interval := time.Duration(float64(time.Second) / ratePerSec)
if interval < time.Microsecond {
interval = time.Microsecond
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-b.done:
return
case <-ticker.C:
select {
case ch := <-b.queue:
close(ch) // 唤醒排在最前面的那个请求
default:
// 队列是空的,这一滴白漏了——漏桶就是这样,没水也照漏
}
}
}
}
// Acquire 排队等待放行。队满立即返回 ErrBucketFull,不做无限等待。
func (b *ShapingLeakyBucket) Acquire(ctx context.Context) error {
ch := make(chan struct{})
select {
case b.queue <- ch:
default:
return ErrBucketFull
}
select {
case <-ch:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (b *ShapingLeakyBucket) Close() {
b.once.Do(func() { close(b.done) })
}
这个版本里，请求是真的在等。空闲再久，队列也是空的，不会攒出额外容量，来一波请求只能一个一个按 1/rate 的节奏出去。这是它和令牌桶唯一实质性的差别。
代价也很直接：每个被限流的请求都要挂一个 goroutine 等结果，而且定时器精度撑不住太高的 QPS。别指望用它去整形一个 50000 QPS 的入口。
对比表和参数决策
| 令牌桶 | 准入式漏桶 | 整形式漏桶 | |
|---|---|---|---|
| 内部状态 | 剩余令牌数 | 当前水位 | 一个待放行队列 |
| 空闲之后 | 攒满 burst 个令牌 | 水位归零 | 队列为空，无积累 |
| 能否突发 | 能，上限 burst | 能，上限 capacity | 不能，严格按 1/rate 间隔 |
| 调用方等待 | 不等待 | 不等待 | 等待 |
| 实现复杂度 | 一把锁 | 一把锁 | 后台 goroutine + channel |
| 高 QPS 适应性 | 好 | 好 | 差，受定时器精度限制 |
| 典型场景 | API 网关、允许瞬时峰值的下游 | 同左 | 严格平滑输出的链路 |
burst 不是越大越好，窗口边界会咬人
回到最开始那个第三方 100 QPS。如果把桶设成 NewTokenBucket(100, 100)，冷启动瞬间就能发出 100 个请求。如果对方用的是固定窗口计数（大部分第三方都是），假设它们的窗口从 t=0.5s 开始计时，我们会在 0.5 秒里发出 150 个，照样超。
稳妥的取值是 burst <= rate，甚至 burst = rate / 2。桶容量应该按「下游在任意一个 1 秒窗口内最多能扛住多少」来定，不是按「我们最多能攒多少」来定。
同一件事换个角度：burst 保护的是你的重试逻辑，不是你的正常流量。 正常流量应该被 rate 卡住，burst 只用来吸收调度抖动带来的瞬时堆积。
单机限流在多副本下会失效
假设服务部署了 3 个副本，每个副本各跑一个 NewTokenBucket(100, 100)，加起来就是 300 QPS——直接超了对方限额的 3 倍。两条路：
- 静态切分：每个副本分
rate/n。简单，但扩缩容那一刻配额是错的，HPA 从 3 个扩到 10 个会有几秒钟的过冲。 - 集中式：用 Redis 存桶状态，Lua 脚本保证原子性。但每次请求都要打一次 Redis，多加 0.5~1ms 延迟。
Redis 版本的 Lua（关键是 now 用 Redis 自己的 TIME，别信应用服务器的时钟，NTP 校时那一下会让你多补或少补令牌）：
-- KEYS[1]: 桶的 key
-- ARGV[1]: rate 每秒补充令牌数
-- ARGV[2]: burst 桶容量
-- ARGV[3]: 本次要取的令牌数
local rate = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local want = tonumber(ARGV[3])
local t = redis.call('TIME') -- {秒, 微秒},脚本内取到的是固定值
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local data = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(data[1]) or burst -- 首次访问,桶是满的
local ts = tonumber(data[2]) or now
local elapsed = math.max(0, now - ts) / 1000 -- 时钟回拨时按 0 处理
tokens = math.min(burst, tokens + elapsed * rate)
local allowed = 0
if tokens >= want then
tokens = tokens - want
allowed = 1
end
redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)
-- 桶从空到满需要 burst/rate 秒,超过这个时间 key 就没必要留着了
redis.call('PEXPIRE', KEYS[1], math.ceil(burst / rate * 1000) + 1000)
return allowed
顺带说一句：Redis 版本的限流精度受 RTT 影响，脚本执行本身是原子的，但从应用发出请求到 Redis 执行完这段时间里，别的副本可能已经抢走了令牌。所以实际放行速率会略低于设定值，这个偏差在 100 QPS 量级大概 1%~3%，高 QPS 下可以忽略。要精确的话只能上分布式协调，不值得。
一定会踩到的边界
golang.org/x/time/rate 的 WaitN 在 n 超过 burst 时直接报错，不是慢等。
lim := rate.NewLimiter(10, 5) // 10 QPS, burst 5
err := lim.WaitN(ctx, 10)
// 立即返回: rate: Wait(n=10) exceeds limiter's burst 5
很多人以为它会排队等到攒够 10 个令牌。不会，token bucket 的 burst 是硬上限，一次要的比桶还大就永远满足不了。我的 AllowN 里也做了同样的处理，直接返回 false 而不是让调用方空转。
burst = 0 会让限流器永远拒绝。 rate.NewLimiter(rate.Limit(100), 0) 之后任何 Allow() 都返回 false。构造时加个断言，别等到线上才发现。
被拒绝之后不要立刻重试。 这是 429 风暴的真正成因。正确做法是返回 429 加 Retry-After 头，客户端按指数退避。如果服务端自己就是客户端，用 rate.Limiter.Wait(ctx) 阻塞等而不是轮询 Allow()——Wait 内部是按需休眠，不是自旋。
动态调速率要小心已经有请求在等。 x/time/rate 的 SetLimit / SetBurst 是线程安全的，但把 burst 调小的瞬间，正在等待的请求可能因为 n 大于新的 burst 而直接失败。控制面下发新配置时最好加个 5 秒的渐变窗口。
单桶的锁在极端场景下会变成瓶颈。 按 key 分片能缓解，但注意分片之后每个桶的 rate 该设多少：如果所有分片共享一份总配额，每个分片只能分到 1/N；如果每个 key 独立限流，分片只是 hash 落点。搞混了会把总速率放大 N 倍。
最后给个判断标准
选之前先回答一个问题：你要的是「把多余的请求拒掉」，还是「让请求平滑地排队」？
拒掉用令牌桶——实现简单、无等待、高 QPS 扛得住，代价是被拒绝的调用方要自己处理 429 和退避。
排队用整形式漏桶——输出端绝对平滑，代价是每个请求都要挂一个等待的 goroutine，而且速率上限被定时器精度卡在几千 QPS。
如果服务是单进程、又没有非要用 Redis 的分布式需求，直接上 golang.org/x/time/rate，别自己写。它的实现比上面这段多处理了「等待队列里的请求按顺序唤醒」这类细节，自己造轮子大概率会在并发压力下翻车。真到了 bucket 上的锁成为瓶颈那天，再考虑按 key 分片或者上本地 + 中心的两级限流。