4、漏桶实现：基于 time.Ticker 的漏桶实现、自定义漏桶结构体

漏桶算法（Leaky Bucket）是限流领域最直观的流量整形模型之一。与令牌桶的“积攒令牌再消费”不同，漏桶强制请求以恒定速率流出，超出桶容量的请求直接丢弃。在 Go 语言中，利用 time.Ticker 实现漏桶是最贴近底层系统时钟的工程方案。

4.1 漏桶核心模型与结构体设计

自定义漏桶结构体需要封装三个核心属性：桶容量（burst）、恒定流出速率（rate）、以及用于驱动周期性流出的 time.Ticker。同时需要记录当前桶中的“水位”（pending），即等待处理的请求数。

4.2 基于 time.Ticker 的漏桶实现

以下代码展示了完整的自定义漏桶结构体及其核心方法：NewLeakyBucket、Allow、StartDrain、Stop。

package leakybucket

import (
    "sync"
    "sync/atomic"
    "time"
)

// LeakyBucket 漏桶限流器
type LeakyBucket struct {
    capacity int64           // 桶容量
    rate     time.Duration   // 漏出间隔
    ticker   *time.Ticker    // 周期性漏出定时器
    pending  int64           // 当前等待数（原子操作）
    mu       sync.Mutex      // 保护 ticker 启停
    done     chan struct{}   // 通知 drain goroutine 退出
}

// NewLeakyBucket 创建一个漏桶限流器
// capacity: 桶容量；rate: 每次漏出的时间间隔
func NewLeakyBucket(capacity int64, rate time.Duration) *LeakyBucket {
    lb := &LeakyBucket{
        capacity: capacity,
        rate:     rate,
        done:     make(chan struct{}),
    }
    return lb
}

// StartDrain 启动后台漏出协程
func (lb *LeakyBucket) StartDrain() {
    lb.mu.Lock()
    defer lb.mu.Unlock()
    if lb.ticker != nil {
        return // 已经启动
    }
    lb.ticker = time.NewTicker(lb.rate)
    go func() {
        for {
            select {
            case <-lb.ticker.C:
                // 每次 tick 漏出一个请求
                for {
                    current := atomic.LoadInt64(&lb.pending)
                    if current == 0 {
                        break
                    }
                    if atomic.CompareAndSwapInt64(&lb.pending, current, current-1) {
                        break
                    }
                }
            case <-lb.done:
                return
            }
        }
    }()
}

// Allow 尝试向桶中放入一个请求
// 返回 true 表示放入成功（请求被接受），false 表示桶满（请求被丢弃）
func (lb *LeakyBucket) Allow() bool {
    for {
        current := atomic.LoadInt64(&lb.pending)
        if current >= lb.capacity {
            return false // 桶满，拒绝
        }
        if atomic.CompareAndSwapInt64(&lb.pending, current, current+1) {
            return true // 放入成功
        }
        // CAS 失败，重试
    }
}

// Stop 停止漏桶，关闭 ticker 和 drain 协程
func (lb *LeakyBucket) Stop() {
    lb.mu.Lock()
    defer lb.mu.Unlock()
    if lb.ticker != nil {
        lb.ticker.Stop()
        lb.ticker = nil
        close(lb.done)
    }
}

4.3 漏桶工作流程（内联 SVG）

下图展示了漏桶从请求入桶到恒定速率漏出的完整生命周期：

4.4 使用示例与边界场景

以下代码演示了如何创建漏桶、启动 drain 协程，并在 HTTP 处理函数中使用：

func main() {
    // 桶容量 10，每 100ms 漏出一个请求（即 10 QPS）
    lb := leakybucket.NewLeakyBucket(10, 100*time.Millisecond)
    lb.StartDrain()
    defer lb.Stop()

    http.HandleFunc("/api", func(w http.ResponseWriter, r *http.Request) {
        if !lb.Allow() {
            http.Error(w, "429 Too Many Requests", http.StatusTooManyRequests)
            return
        }
        w.Write([]byte("OK"))
    })

    log.Fatal(http.ListenAndServe(":8080", nil))
}

4.5 漏桶 vs 令牌桶：核心差异