Gin限流与熔断机制

限流和熔断是保障服务稳定性的关键机制，防止系统被过载请求压垮。

限流算法

固定窗口算法

type FixedWindowLimiter struct {
    limit    int
    window   time.Duration
    counter  int
    lastTime time.Time
    mu       sync.Mutex
}

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

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

    if l.counter < l.limit {
        l.counter++
        return true
    }
    return false
}

func RateLimitMiddleware(limit int, window time.Duration) gin.HandlerFunc {
    limiter := &FixedWindowLimiter{limit: limit, window: window}

    return func(c *gin.Context) {
        if !limiter.Allow() {
            c.JSON(429, gin.H{"error": "rate limit exceeded"})
            c.Abort()
            return
        }
        c.Next()
    }
}

滑动窗口算法

type SlidingWindowLimiter struct {
    limit    int
    window   time.Duration
    requests []time.Time
    mu       sync.Mutex
}

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

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

    // 清理过期请求
    validRequests := []time.Time{}
    for _, t := range l.requests {
        if t.After(windowStart) {
            validRequests = append(validRequests, t)
        }
    }
    l.requests = validRequests

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

令牌桶算法

type TokenBucket struct {
    capacity  int
    tokens    int
    rate      float64  // 每秒添加令牌数
    lastTime  time.Time
    mu        sync.Mutex
}

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

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

    // 添加令牌
    tb.tokens += int(elapsed * tb.rate)
    if tb.tokens > tb.capacity {
        tb.tokens = tb.capacity
    }

    if tb.tokens > 0 {
        tb.tokens--
        return true
    }
    return false
}

漏桶算法

type LeakyBucket struct {
    capacity  int
    remaining int
    rate      float64  // 每秒流出请求数
    lastTime  time.Time
    mu        sync.Mutex
}

func (lb *LeakyBucket) Allow() bool {
    lb.mu.Lock()
    defer lb.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(lb.lastTime).Seconds()
    lb.lastTime = now

    // 流出请求
    outflow := int(elapsed * lb.rate)
    lb.remaining -= outflow
    if lb.remaining < 0 {
        lb.remaining = 0
    }

    if lb.remaining < lb.capacity {
        lb.remaining++
        return true
    }
    return false
}

分布式限流

Redis实现

func RedisRateLimitMiddleware(rdb *redis.Client, key string, limit int, window int) gin.HandlerFunc {
    return func(c *gin.Context) {
        ctx := c.Request.Context()

        // Lua脚本保证原子性
        script := `
            local current = redis.call('INCR', KEYS[1])
            if current == 1 then
                redis.call('EXPIRE', KEYS[1], ARGV[1])
            end
            return current
        `

        result, err := rdb.Eval(ctx, script, []string{key}, window).Int()
        if err != nil {
            c.JSON(500, gin.H{"error": "rate limit check failed"})
            c.Abort()
            return
        }

        if result > limit {
            c.JSON(429, gin.H{"error": "rate limit exceeded"})
            c.Abort()
            return
        }

        c.Next()
    }
}

// 使用：按用户ID限流
r.Use(func(c *gin.Context) {
    userID := c.GetString("userID")
    RedisRateLimitMiddleware(rdb, "rate:"+userID, 100, 60)(c)
})

熔断机制

熔断器实现

type CircuitBreaker struct {
    failures     int
    threshold    int
    state        string  // "closed", "open", "half-open"
    lastFailure  time.Time
    recoveryTime time.Duration
    mu           sync.Mutex
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mu.Lock()

    switch cb.state {
    case "open":
        if time.Since(cb.lastFailure) > cb.recoveryTime {
            cb.state = "half-open"
        } else {
            cb.mu.Unlock()
            return fmt.Errorf("circuit breaker is open")
        }
    }

    cb.mu.Unlock()

    err := fn()

    cb.mu.Lock()
    defer cb.mu.Unlock()

    if err != nil {
        cb.failures++
        cb.lastFailure = time.Now()

        if cb.failures >= cb.threshold {
            cb.state = "open"
        }
        return err
    }

    cb.failures = 0
    cb.state = "closed"
    return nil
}

熔断中间件

func CircuitBreakerMiddleware(service string, threshold int, recovery time.Duration) gin.HandlerFunc {
    cb := &CircuitBreaker{
        threshold:    threshold,
        recoveryTime: recovery,
        state:        "closed",
    }

    return func(c *gin.Context) {
        if cb.state == "open" {
            c.JSON(503, gin.H{
                "error":   "service unavailable",
                "service": service,
            })
            c.Abort()
            return
        }

        c.Next()

        if c.Writer.Status() >= 500 {
            cb.failures++
            if cb.failures >= cb.threshold {
                cb.state = "open"
                cb.lastFailure = time.Now()
            }
        } else {
            cb.failures = 0
        }
    }
}

算法对比

分级限流策略

// 多级限流
func MultiLevelRateLimit() gin.HandlerFunc {
    globalLimiter := NewTokenBucket(10000, 1000)  // 全局10000/秒
    userLimiter := make(map[string]*TokenBucket) // 用户级100/秒

    return func(c *gin.Context) {
        // 1. 全局限流
        if !globalLimiter.Allow() {
            c.JSON(429, gin.H{"error": "global rate limit"})
            c.Abort()
            return
        }

        // 2. 用户限流
        userID := c.GetString("userID")
        limiter, ok := userLimiter[userID]
        if !ok {
            limiter = NewTokenBucket(100, 10)
            userLimiter[userID] = limiter
        }

        if !limiter.Allow() {
            c.JSON(429, gin.H{"error": "user rate limit"})
            c.Abort()
            return
        }

        c.Next()
    }
}

注意：分布式限流需使用Redis保证原子性，Lua脚本是最佳实践。

要点总结

• 令牌桶允许突发流量，适合API限流
• 漏桶恒定流出，适合流量整形
• 滑动窗口精确统计，无边界问题
• Redis Lua脚本保证分布式限流原子性
• 熔断器保护下游服务，避免级联故障
• 多级限流：全局→用户→接口