Go与限流

00:00

3 min Intermediate 2026/6/14

限流与熔断

概述

限流是控制单位时间内请求数量的技术，防止系统因过载而崩溃。当流量超过系统处理能力时，限流会拒绝多余的请求，保护后端服务。熔断则是在检测到服务异常时主动断开调用，避免故障扩散。Go 标准库和社区提供了多种限流和熔断工具。

基础概念

在开始编码之前，需要理解限流的几个核心概念：

限流（Rate Limiting）：限制单位时间内的请求数量，超出限制的请求被拒绝或排队。
熔断（Circuit Breaking）：当下游服务异常率超过阈值时，自动断开调用，直接返回错误，避免浪费资源等待超时。
令牌桶（Token Bucket）：限流算法之一，以固定速率向桶中放入令牌，请求需要消耗令牌，桶满时多余令牌丢弃。
漏桶（Leaky Bucket）：限流算法之一，请求进入桶中排队，以固定速率处理。
滑动窗口：限流算法之一，统计最近一个时间窗口内的请求数量。

快速上手

使用 Go 扩展库的令牌桶限流器：

go get golang.org/x/time/rate

package main

import (
    "fmt"
    "golang.org/x/time/rate"
)

func main() {
    // 创建限流器：每秒产生 100 个令牌，桶容量 10
    limiter := rate.NewLimiter(100, 10)

    // 尝试获取令牌
    if limiter.Allow() {
        fmt.Println("请求通过")
    } else {
        fmt.Println("请求被限流")
    }
}

详细用法

1. 令牌桶限流（rate.Limiter）

import "golang.org/x/time/rate"

// 创建限流器
// 参数1：每秒产生的令牌数（rate）
// 参数2：桶容量（burst），允许瞬间通过的最多请求数
limiter := rate.NewLimiter(100, 10)

// 方式1：Allow - 立即返回是否允许
if limiter.Allow() {
    // 处理请求
} else {
    // 返回 429 Too Many Requests
}

// 方式2：Wait - 阻塞等待直到获取令牌
err := limiter.Wait(context.Background())
if err != nil {
    // Context 被取消
}
// 处理请求

// 方式3：Reserve - 预留令牌，返回需要等待的时间
reservation := limiter.Reserve()
if !reservation.OK() {
    // 令牌数超过桶容量，无法满足
    return
}
time.Sleep(reservation.Delay()) // 等待预留的延迟时间
// 处理请求

// 动态调整速率
limiter.SetRate(200)    // 调整为每秒 200 个
limiter.SetBurst(20)    // 调整桶容量为 20

2. HTTP 中间件限流

将限流器集成到 HTTP 服务中：

func RateLimitMiddleware(limiter *rate.Limiter) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if !limiter.Allow() {
                http.Error(w, "请求过于频繁，请稍后再试", http.StatusTooManyRequests)
                return
            }
            next.ServeHTTP(w, r)
        })
    }
}

// 使用
limiter := rate.NewLimiter(100, 10)
handler := RateLimitMiddleware(limiter)(mux)
http.ListenAndServe(":8080", handler)

3. 按客户端限流

为每个客户端（IP 或用户 ID）维护独立的限流器：

type IPRateLimiter struct {
    limiters sync.Map
    rate     rate.Limit
    burst    int
}

func NewIPRateLimiter(r rate.Limit, burst int) *IPRateLimiter {
    return &IPRateLimiter{rate: r, burst: burst}
}

func (l *IPRateLimiter) GetLimiter(ip string) *rate.Limiter {
    if limiter, ok := l.limiters.Load(ip); ok {
        return limiter.(*rate.Limiter)
    }
    limiter := rate.NewLimiter(l.rate, l.burst)
    l.limiters.Store(ip, limiter)
    return limiter
}

// 中间件
func (l *IPRateLimiter) Middleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ip := r.RemoteAddr
        if !l.GetLimiter(ip).Allow() {
            http.Error(w, "请求过于频繁", http.StatusTooManyRequests)
            return
        }
        next.ServeHTTP(w, r)
    })
}

4. 滑动窗口限流

使用 Redis 实现分布式滑动窗口限流：

func SlidingWindowLimit(rdb *redis.Client, key string, limit int, window time.Duration) (bool, error) {
    ctx := context.Background()
    now := time.Now().UnixNano()
    windowStart := now - window.Nanoseconds()

    pipe := rdb.Pipeline()
    // 移除窗口外的记录
    pipe.ZRemRangeByScore(ctx, key, "0", fmt.Sprintf("%d", windowStart))
    // 添加当前请求
    pipe.ZAdd(ctx, key, redis.Z{Score: float64(now), Member: now})
    // 统计窗口内请求数
    countCmd := pipe.ZCard(ctx, key)
    // 设置过期时间
    pipe.Expire(ctx, key, window)

    _, err := pipe.Exec(ctx)
    if err != nil {
        return false, err
    }

    return countCmd.Val() <= int64(limit), nil
}

5. 熔断器

使用 github.com/sony/gobreaker 实现熔断：

go get github.com/sony/gobreaker

import "github.com/sony/gobreaker"

// 创建熔断器
cb := gobreaker.NewCircuitBreaker(gobreaker.Settings{
    Name:        "order-service",
    MaxRequests: 5,                    // 半开状态下允许的测试请求数
    Interval:    60 * time.Second,     // 统计窗口
    Timeout:     30 * time.Second,     // 熔断后等待多久尝试恢复
    ReadyToTrip: func(counts gobreaker.Counts) bool {
        // 连续 5 次失败则熔断
        return counts.ConsecutiveFailures > 5
    },
})

// 使用熔断器包装调用
result, err := cb.Execute(func() (interface{}, error) {
    resp, err := http.Get("http://order-service/api/orders")
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 500 {
        return nil, fmt.Errorf("服务端错误: %d", resp.StatusCode)
    }
    return io.ReadAll(resp.Body), nil
})

if err != nil {
    // 熔断器打开时，err 为 gobreaker.ErrOpenCircuit
    if err == gobreaker.ErrOpenCircuit {
        fmt.Println("熔断器已打开，快速失败")
    }
}

6. 并发限流

使用信号量限制并发数：

go get golang.org/x/sync/semaphore

import "golang.org/x/sync/semaphore"

// 最多 10 个并发
sem := semaphore.NewWeighted(10)

func handler(w http.ResponseWriter, r *http.Request) {
    // 获取信号量
    if err := sem.Acquire(r.Context(), 1); err != nil {
        http.Error(w, "服务器繁忙", http.StatusServiceUnavailable)
        return
    }
    defer sem.Release(1)

    // 处理请求
    processRequest(w, r)
}

常见场景

场景一：API 网关限流

// 全局限流 + 按用户限流
globalLimiter := rate.NewLimiter(1000, 50)   // 全局每秒 1000
userLimiters := NewIPRateLimiter(10, 5)       // 每用户每秒 10

func GatewayMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 全局限流
        if !globalLimiter.Allow() {
            http.Error(w, "系统繁忙", http.StatusTooManyRequests)
            return
        }
        // 用户限流
        if !userLimiters.GetLimiter(r.RemoteAddr).Allow() {
            http.Error(w, "请求过于频繁", http.StatusTooManyRequests)
            return
        }
        next.ServeHTTP(w, r)
    })
}

场景二：下游服务熔断保护

var userCB *gobreaker.CircuitBreaker

func init() {
    userCB = gobreaker.NewCircuitBreaker(gobreaker.Settings{
        Name:    "user-service",
        Timeout: 30 * time.Second,
        ReadyToTrip: func(c gobreaker.Counts) bool {
            return c.ConsecutiveFailures > 3
        },
    })
}

func GetUser(ctx context.Context, id string) (*User, error) {
    result, err := userCB.Execute(func() (interface{}, error) {
        return userClient.Get(ctx, id)
    })
    if err != nil {
        return nil, err // 熔断时快速失败
    }
    return result.(*User), nil
}

注意事项与常见错误

限流粒度选择：全局限流保护系统，用户级限流防止单个用户滥用。两者通常组合使用。
429 响应要包含重试信息：返回 Retry-After 头，告诉客户端多久后可以重试：

w.Header().Set("Retry-After", "1")
http.Error(w, "请求过于频繁", http.StatusTooManyRequests)

限流器内存泄漏：按客户端限流时，如果客户端数量无限增长，sync.Map 会持续膨胀。需要定期清理不活跃的限流器。
分布式限流：单机限流在多实例部署时无法全局生效。需要使用 Redis 等共享存储实现分布式限流。
熔断器不是万能的：熔断器适合保护下游服务调用，不适合保护内部逻辑。
令牌桶 vs 漏桶：令牌桶允许突发流量（burst），漏桶强制匀速。大多数场景用令牌桶更合适。

进阶用法

自适应限流

根据系统负载动态调整限流阈值：

type AdaptiveLimiter struct {
    limiter *rate.Limiter
    mu      sync.Mutex
}

func (l *AdaptiveLimiter) Adjust(cpuUsage float64) {
    l.mu.Lock()
    defer l.mu.Unlock()

    // CPU 使用率低时提高限流阈值
    if cpuUsage < 0.5 {
        l.limiter.SetRate(l.limiter.Limit() * 1.1)
    }
    // CPU 使用率高时降低限流阈值
    if cpuUsage > 0.8 {
        l.limiter.SetRate(l.limiter.Limit() * 0.9)
    }
}

限流响应格式

返回标准的限流响应：

func RateLimitResponse(w http.ResponseWriter, retryAfter int) {
    w.Header().Set("Content-Type", "application/json")
    w.Header().Set("Retry-After", fmt.Sprintf("%d", retryAfter))
    w.WriteHeader(http.StatusTooManyRequests)
    json.NewEncoder(w).Encode(map[string]interface{}{
        "error":       "too_many_requests",
        "message":     "请求过于频繁，请稍后再试",
        "retry_after": retryAfter,
    })
}

Go与限流

概述

基础概念

快速上手

详细用法

1. 令牌桶限流（rate.Limiter）

2. HTTP 中间件限流

3. 按客户端限流

4. 滑动窗口限流

5. 熔断器

6. 并发限流

常见场景

场景一：API 网关限流

场景二：下游服务熔断保护

注意事项与常见错误

进阶用法

自适应限流

限流响应格式

知识检测

学习进度

Go与限流

知识检测

学习进度

学习推荐