限频的六种姿势
从原理到突发流量

核心思想

把时间切成长度相等的"桶"（比如 1 秒），每进入一个桶时把计数器清零， 之后每个请求 +1，超过阈值就拒绝。

实现只需要两个变量：windowStart 和 counter。

实现（Go）

type FixedWindow struct {
    mu          sync.Mutex
    window      time.Duration
    limit       int
    counter     int
    windowStart time.Time
}

func NewFixedWindow(window time.Duration, limit int) *FixedWindow {
    return &FixedWindow{
        window:      window,
        limit:       limit,
        windowStart: time.Now(),
    }
}

func (fw *FixedWindow) Allow() bool {
    fw.mu.Lock()
    defer fw.mu.Unlock()

    now := time.Now()
    if now.Sub(fw.windowStart) >= fw.window {
        fw.windowStart = now
        fw.counter     = 0
    }

    if fw.counter >= fw.limit {
        return false
    }
    fw.counter++
    return true
}

核心思想

用一条队列（或数组）保存每一次成功请求的时间戳。 每次请求来时，先把队头里"超出窗口"的过期时间戳弹掉， 再看队列长度是否超限。

实现（Go）

type SlidingLog struct {
    mu     sync.Mutex
    window time.Duration
    limit  int
    log    []time.Time   // 每次成功请求的时间戳
}

func (l *SlidingLog) Allow() bool {
    l.mu.Lock()
    defer l.mu.Unlock()

    now := time.Now()
    cutoff := now.Add(-l.window)

    // 弹出已过期的旧时间戳
    i := 0
    for i < len(l.log) && l.log[i].Before(cutoff) {
        i++
    }
    l.log = l.log[i:]

    if len(l.log) >= l.limit {
        return false
    }
    l.log = append(l.log, now)
    return true
}

核心思想

把时间切成固定窗口，但统计时用上一个完整窗口的计数加权估算"过去"。 这是 Cloudflare 在工程上大量使用的方案——精度够用、内存 O(1)。

公式：
estimate = prevCount × (1 - elapsed/window) + currCount

实现（Go）

type SlidingCounter struct {
    mu          sync.Mutex
    window      time.Duration
    limit       int
    currCount   int
    prevCount   int
    currStart   time.Time
}

func (s *SlidingCounter) Allow() bool {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(s.currStart)

    if elapsed >= s.window {
        s.prevCount = s.currCount
        s.currCount = 0
        s.currStart = now
        elapsed = 0
    }

    weight := 1 - float64(elapsed)/float64(s.window)
    estimate := float64(s.prevCount)*weight + float64(s.currCount)

    if int(estimate) >= s.limit {
        return false
    }
    s.currCount++
    return true
}

核心思想

维护一个容量为 capacity 的桶，以速率 rate 匀速往桶里放令牌， 桶满则新令牌丢弃。每个请求消耗 1 个令牌，没令牌就拒绝。

两个旋钮：rate（平均速率）+ capacity（桶容量 = 突发上限）。

实现（Go）

type TokenBucket struct {
    mu        sync.Mutex
    rate      float64   // 每秒放几个令牌
    capacity  int        // 桶最大容量
    tokens    float64   // 当前令牌数
    lastToken time.Time  // 上次补充时间
}

func (tb *TokenBucket) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(tb.lastToken).Seconds()

    // 补充令牌：最多补到 capacity
    tb.tokens = math.Min(
        float64(tb.capacity),
        tb.tokens + elapsed*tb.rate,
    )
    tb.lastToken = now

    if tb.tokens < 1 {
        return false
    }
    tb.tokens--
    return true
}

核心思想

请求进入桶中排队，桶以固定速率处理它们。 桶满时新请求被拒绝。它和令牌桶是镜像关系： 一个是"攒够了再发"，一个是"匀速发出"。

实现（Go · 单机版）

type LeakyBucket struct {
    mu       sync.Mutex
    rate     time.Duration  // 每隔多久漏一个
    capacity int                 // 桶容量
    queue    chan struct{}
    stop     chan struct{}
}

func NewLeakyBucket(rate time.Duration, cap int) *LeakyBucket {
    lb := &LeakyBucket{
        rate:     rate,
        capacity: cap,
        queue:    make(chan struct{}, cap),
        stop:     make(chan struct{}),
    }
    go lb.leak()
    return lb
}

func (lb *LeakyBucket) leak() {
    t := time.NewTicker(lb.rate)
    defer t.Stop()
    for {
        select {
        case <-t.C:
            <-lb.queue   // 漏一个
        case <-lb.stop:
            return
        }
    }
}

func (lb *LeakyBucket) Allow() bool {
    select {
    case lb.queue <- struct{}{}:
        return true
    default:
        return false   // 桶满，拒绝
    }
}

核心思想

上面所有算法限制的都是"单位时间内的请求数"。但实际生产中更危险的是： 同一瞬间有 1000 个慢查询堆积，每个 10 秒，QPS 才 100，但服务一样挂。

并发限频用一个信号量 (Semaphore) 限制同时处理的请求数。 它不关心"来过几个"，只关心"现在有几个人在"。

实现（Go）

type ConcurrencyLimiter struct {
    sem chan struct{}
}

func New(limit int) *ConcurrencyLimiter {
    return &ConcurrencyLimiter{
        sem: make(chan struct{}, limit),
    }
}

func (c *ConcurrencyLimiter) Do(ctx context.Context, fn func() error) error {
    select {
    case c.sem <- struct{}{}:
        // 拿到许可
    case <-ctx.Done():
        return ctx.Err()   // 超时直接拒
    }
    defer func() { <-c.sem }()

    return fn()
}

// 用法
limiter := New(100)
for _, req := range requests {
    go limiter.Do(ctx, req.Handle)
}