架构师核心能力：限流的底层原理解析

👉目录

1 限流的目的

2 限流算法的实现

3 限流的实现方式

4 限流策略

5 限流考虑的因素

02

// 这是一个简单的实现案例package main
import (  "fmt"  "sync"  "time")
// FixedWindowCounter 结构体实现固定窗口计数器限流算法。// mu 用于同步访问，保证并发安全。// count 记录当前时间窗口内的请求数量。// limit 是时间窗口内允许的最大请求数量。// window 记录当前时间窗口的开始时间。// duration 是时间窗口的持续时间。type FixedWindowCounter struct {  mu        sync.Mutex  count     int  limit     int  window    time.Time  duration  time.Duration}
// NewFixedWindowCounter 构造函数初始化 FixedWindowCounter 实例。// limit 参数定义了每个时间窗口内允许的请求数量。// duration 参数定义了时间窗口的大小。func NewFixedWindowCounter(limit int, duration time.Duration) *FixedWindowCounter {  return &FixedWindowCounter{    limit:   limit,    window:  time.Now(),   // 设置当前时间作为窗口的开始时间。    duration: duration,    // 设置时间窗口的持续时间。  }}
// Allow 方法用于判断当前请求是否被允许。// 首先通过互斥锁保证方法的原子性。func (f *FixedWindowCounter) Allow() bool {  f.mu.Lock()  defer f.mu.Unlock()
  now := time.Now() // 获取当前时间。
  // 如果当前时间超过了窗口的结束时间，重置计数器和窗口开始时间。  if now.After(f.window.Add(f.duration)) {    f.count = 0    f.window = now  }
  // 如果当前计数小于限制，则增加计数并允许请求。  if f.count < f.limit {    f.count++    return true  }  // 如果计数达到限制，则拒绝请求。  return false}
// main 函数是程序的入口点。func main() {  // 创建一个新的限流器，设置每分钟（time.Minute）只允许10个请求。  limiter := NewFixedWindowCounter(10, time.Minute)
  // 模拟15个请求，观察限流效果。  for i := 0; i < 15; i++ {    if limiter.Allow() {      fmt.Println("Request", i+1, "allowed")    } else {      fmt.Println("Request", i+1, "rejected")    }  }}

// 这是一个简单的实现案例，这个代码示例仅用于说明滑动窗口限流算法的逻辑，并非完整的工作代码。
package main
import (  "fmt"  "sync"  "time")
// SlidingWindowLimiter 结构体实现滑动窗口限流算法。type SlidingWindowLimiter struct {  mutex       sync.Mutex  counters    []int  limit       int  windowStart time.Time  windowDuration time.Duration  interval    time.Duration}
// NewSlidingWindowLimiter 构造函数初始化 SlidingWindowLimiter 实例。func NewSlidingWindowLimiter(limit int, windowDuration time.Duration, interval time.Duration) *SlidingWindowLimiter {  buckets := int(windowDuration / interval)  return &SlidingWindowLimiter{    counters:    make([]int, buckets),    limit:       limit,    windowStart: time.Now(),    windowDuration: windowDuration,    interval:    interval,  }}
// Allow 方法用于判断当前请求是否被允许，并实现滑动窗口的逻辑。func (s *SlidingWindowLimiter) Allow() bool {  s.mutex.Lock()  defer s.mutex.Unlock()
  // 检查是否需要滑动窗口  if time.Since(s.windowStart) > s.windowDuration {    s.slideWindow()  }
  now := time.Now()  index := int((now.UnixNano() - s.windowStart.UnixNano()) / s.interval.Nanoseconds()) % len(s.counters)
  if s.counters[index] < s.limit {    s.counters[index]++    return true  }  return false}
// slideWindow 方法实现滑动窗口逻辑，移除最旧的时间段并重置计数器。func (s *SlidingWindowLimiter) slideWindow() {  // 滑动窗口，忽略最旧的时间段  copy(s.counters, s.counters[1:])  // 重置最后一个时间段的计数器  s.counters[len(s.counters)-1] = 0  // 更新窗口开始时间  s.windowStart = time.Now()}
// main 函数是程序的入口点。func main() {  limiter := NewSlidingWindowLimiter(1, time.Second, 10*time.Millisecond)
  for i := 0; i < 100; i++ {    if limiter.Allow() {      fmt.Println("Request", i+1, "allowed")    } else {      fmt.Println("Request", i+1, "rejected")    }  }}

// 这是一个简单的实现案例，这个代码示例仅用于说明漏桶算法的基本逻辑，并非完整的工作代码。
package main
import (    "fmt"    "time")
// LeakyBucket 结构体，包含请求队列type LeakyBucket struct {    queue chan struct{} // 请求队列}
// NewLeakyBucket 创建一个新的漏桶实例func NewLeakyBucket(capacity int) *LeakyBucket {    return &LeakyBucket{       queue: make(chan struct{}, capacity),    }}
// push 将请求放入队列，如果队列满了，返回 false，表示请求被丢弃func (lb *LeakyBucket) push() bool {    // 如果通道可以发送，请求被接受    select {    case lb.queue <- struct{}{}:       return true    default:       return false    }}
// process 从队列中取出请求并模拟处理过程func (lb *LeakyBucket) process() {    for range lb.queue { // 使用 range 来持续接收队列中的请求       fmt.Println("Request processed at", time.Now().Format("2006-01-02 15:04:05"))       time.Sleep(100 * time.Millisecond) // 模拟请求处理时间    }}
func main() {    lb := NewLeakyBucket(5) // 创建一个容量为5的漏桶
    // 启动请求处理循环    go lb.process()
    // 模拟请求    for i := 0; i < 10; i++ {       accepted := lb.push()       if accepted {          fmt.Printf("Request %d accepted at %v\n", i+1, time.Now().Format("2006-01-02 15:04:05"))       } else {          fmt.Printf("Request %d rejected at %v\n", i+1, time.Now().Format("2006-01-02 15:04:05"))       }    }    time.Sleep(2 * time.Second)}

// 这是一个简单的实现案例，这个代码示例仅用于说明令牌桶算法的基本逻辑，并非完整的工作代码。package main
import (    "fmt"    "sync"    "time")
// TokenBucket 结构体实现令牌桶限流算法。// - mu 用于同步访问，保证并发安全。// - capacity 定义桶的容量，即桶中最多可以存放的令牌数。// - tokens 表示桶中当前的令牌数。// - refillRate 是令牌的填充速率，表示每秒向桶中添加的令牌数。// - lastRefill 记录上次填充令牌的时间。type TokenBucket struct {    mu         sync.Mutex    capacity   int    tokens     int    refillRate float64    lastRefill time.Time}
// NewTokenBucket 构造函数初始化 TokenBucket 实例。// - capacity 参数定义了桶的容量。// - refillRate 参数定义了每秒向桶中添加的令牌数。func NewTokenBucket(capacity int, refillRate float64) *TokenBucket {    // 初始化时桶被填满，tokens 和 capacity 相等。    // lastRefill 设置为当前时间。    return &TokenBucket{       capacity:   capacity,       tokens:     capacity,       refillRate: refillRate,       lastRefill: time.Now(),    }}
// Allow 方法用于判断当前请求是否被允许。func (t *TokenBucket) Allow() bool {    t.mu.Lock() // 进入临界区，确保操作的原子性。    defer t.mu.Unlock()
    now := time.Now() // 获取当前时间。
    // 计算自上次填充以来经过的秒数，并转换为float64类型。    timeElapsed := float64(now.Unix() - t.lastRefill.Unix())
    // 根据 refillRate 计算应该添加的令牌数。    tokensToAdd := t.refillRate * timeElapsed
    // 更新令牌数，但不超过桶的容量。    t.tokens += int(tokensToAdd)    if t.tokens > t.capacity {       t.tokens = t.capacity // 确保令牌数不超过桶的容量。    }
    // 如果桶中有令牌，则移除一个令牌并允许请求通过。    if t.tokens > 0 {       t.tokens--         // 移除一个令牌。       t.lastRefill = now // 更新上次填充时间到当前时间。       return true    }
    // 如果桶中无令牌，则请求被拒绝。    return false}
// main 函数是程序的入口点。func main() {    // 创建一个新的令牌桶实例，桶的容量为10，每秒填充2个令牌。    limiter := NewTokenBucket(10, 2)
    // 模拟请求，观察限流效果。    // 循环15次，每次请求判断是否被允许。    for i := 0; i < 15; i++ {       if limiter.Allow() {          fmt.Println("Request", i+1, "allowed")       } else {          fmt.Println("Request", i+1, "rejected")       }    }}

03

// 这是一个伪代码案例，演示实现逻辑package main
import (  "fmt"  "github.com/gin-gonic/gin" // 引入Gin框架，用于构建Web服务器和处理HTTP请求  "net/http"  "sync"                // 引入sync包，用于同步原语，如互斥锁  "time"                 // 引入time包，用于时间相关操作)
// TokenBucket 结构体实现令牌桶限流算法。// 它包含互斥锁mu用于同步访问，capacity代表桶的容量，// tokens表示当前桶中的令牌数，refillRate是令牌的填充速率（每秒），// lastRefill记录上次填充的时间。type TokenBucket struct {  mu        sync.Mutex  capacity  int  tokens    int  refillRate float64  lastRefill time.Time}
// NewTokenBucket 函数创建并初始化一个新的TokenBucket实例。// 它设置桶的容量和填充速率，并将初始令牌数设为容量的值。func NewTokenBucket(capacity int, refillRate float64) *TokenBucket {  return &TokenBucket{    capacity:  capacity,    tokens:    capacity,  // 初始化时桶被填满    refillRate: refillRate,    lastRefill: time.Now(), // 记录创建时的时间作为上次填充时间  }}
// Allow 方法用于检查是否允许通过当前请求。// 它首先获取锁，然后计算自上次填充以来应该添加的令牌数，// 更新桶中的令牌数，但不超过桶的容量。// 如果桶中至少有一个令牌，它将减少一个令牌并返回true，表示请求被允许。// 如果桶为空，则返回false，表示请求被拒绝。func (tb *TokenBucket) Allow() bool {  tb.mu.Lock() // 获取锁，保证操作的原子性  defer tb.mu.Unlock()
  now := time.Now() // 获取当前时间  // 计算自上次填充以来经过的秒数，然后乘以填充速率，得到应添加的令牌数  tokensToAdd := int(tb.refillRate * (now.Sub(tb.lastRefill).Seconds()))  tb.tokens += tokensToAdd // 更新桶中的令牌数  if tb.tokens > tb.capacity {    tb.tokens = tb.capacity // 确保不超过桶的容量  }
  if tb.tokens > 0 {    tb.tokens-- // 处理请求，减少一个令牌    tb.lastRefill = now // 更新上次填充时间为当前时间    return true  }  return false // 如果桶为空，返回false}
// Middleware 函数返回一个Gin中间件，该中间件使用TokenBucket来限流。// 如果TokenBucket的Allow方法返回false，中间件将中断请求处理，// 并返回HTTP状态码429（Too Many Requests）和错误信息。// 如果请求被允许，中间件将调用c.Next()继续执行后续的处理链。func Middleware(tb *TokenBucket) gin.HandlerFunc {  return func(c *gin.Context) {    // 在处理请求之前，调用TokenBucket的Allow方法检查是否允许请求    if !tb.Allow() {      // 如果请求被限流，返回错误信息和状态码      c.JSON(http.StatusTooManyRequests, gin.H{"error": "too many requests"})      c.Abort() // 中断请求处理      return    }    // 如果请求未被限流，继续执行后续的处理链    c.Next()  }}
func main() {  // 创建一个Gin的默认实例，用于Web服务  r := gin.Default()
  // 创建TokenBucket实例，用于限流控制  tb := NewTokenBucket(10, 1.0) // 桶的容量为10，每秒填充1个令牌
  // 使用上面定义的限流中间件  r.Use(Middleware(tb))
  // 定义一个简单的路由，当访问/hello路径时，返回JSON格式的消息  r.GET("/hello", func(c *gin.Context) {    c.JSON(http.StatusOK, gin.H{"message": "hello world"})  })
  // 启动Gin服务器，默认监听在0.0.0.0:8080  r.Run()}

http {    # 定义一个限流区域，使用共享内存存储状态    limit_req_zone $binary_remote_addr zone=mylimit:10m rate=1r/s;
    server {        # 监听80端口        listen 80;
        # 定义一个location块，用于匹配特定的请求路径        location /api/ {            # 应用限流规则            limit_req zone=mylimit burst=5 nodelay;
            # 代理请求到后端服务            proxy_pass http://backend/;        }    }}

04

// RateLimiterV2 结构体增加了阈值设置功能。type RateLimiterV2 struct {    mu     sync.Mutex    tokens int    capacity  int      // 桶的容量，代表最大令牌数    refillRate float64 // 每秒填充的令牌数    limit     int      // 请求处理的阈值，即桶的容量}
// NewRateLimiterV2 创建一个新的RateLimiterV2实例，并设置阈值。func NewRateLimiterV2(capacity int, refillRate float64, limit int) *RateLimiterV2 {    return &RateLimiterV2{        capacity:  capacity,        refillRate: refillRate,        limit:     limit,    }}
// Allow 现在考虑了设置的阈值。func (r *RateLimiterV2) Allow() bool {    r.mu.Lock()    defer r.mu.Unlock()
    // 令牌桶逻辑...    // 如果桶中的令牌数达到或超过阈值，则拒绝请求。    if r.tokens >= r.limit {        return false    }
    // 允许请求逻辑...    return true}

// RouteLimiterMap 是一个映射，存储每个路由路径对应的限流器实例。// 键是路由的字符串表示，值是指向RateLimiterV2类型实例的指针。var RouteLimiterMap = map[string]*RateLimiterV2{}
// SetRateLimiterForRoute 函数为指定的路由设置一个新的限流器。// 它接受路由的路径、桶的容量、每秒填充的令牌数和请求处理的阈值作为参数，// 并创建一个新的RateLimiterV2实例，将其存储在RouteLimiterMap中。func SetRateLimiterForRoute(route string, capacity int, refillRate float64, limit int) {    // 在RouteLimiterMap中为给定的路由创建或更新限流器实例。    RouteLimiterMap[route] = NewRateLimiterV2(capacity, refillRate, limit)}
// MiddlewareWithRoute 函数返回一个Gin中间件处理函数。// 该中间件基于路由名称来应用限流逻辑。func MiddlewareWithRoute(route string) gin.HandlerFunc {    // 返回一个Gin的处理函数，该函数内部封装了限流逻辑。    return func(c *gin.Context) {        // 检查RouteLimiterMap中是否存在对应路由的限流器。        // 如果存在，调用其Allow方法来决定当前请求是否应该被允许。        if !RouteLimiterMap[route].Allow() {            // 如果请求被限流（不允许），返回HTTP 429状态码和错误信息。            c.JSON(http.StatusTooManyRequests, gin.H{"error": "too many requests"})            c.Abort() // 中断请求的进一步处理。            return    // 退出中间件函数。        }        // 如果请求未被限流，调用c.Next继续执行Gin的处理链。        c.Next()    }}

// AllowWithFeedback 提供反馈的请求允许逻辑。func (r *RateLimiterV2) AllowWithFeedback() (bool, string) {    r.mu.Lock()    defer r.mu.Unlock()
    // 令牌桶逻辑...    if r.tokens >= r.limit {        return false, "Too many requests. Please try again later."    }
    // 允许请求逻辑...    r.tokens-- // 移除令牌    return true, ""}
// 使用反馈机制的中间件。func MiddlewareWithFeedback() gin.HandlerFunc {    return func(c *gin.Context) {        allowed, message := RouteLimiterMap["/api/"].AllowWithFeedback()        if !allowed {            c.JSON(http.StatusTooManyRequests, gin.H{"error": message})            c.Abort()            return        }        c.Next()    }}

05

// FairLimiter 结构体实现基于用户ID或IP的公平限流。type FairLimiter struct {    sync.Mutex    limits map[string]*RateLimiterV2 // 为每个用户或IP维护一个独立的限流器}
// NewFairLimiter 创建一个新的FairLimiter实例。func NewFairLimiter(capacity int, refillRate float64) *FairLimiter {    return &FairLimiter{        limits: make(map[string]*RateLimiterV2),    }}
// Allow 根据用户ID或IP决定是否允许请求。func (f *FairLimiter) Allow(userID string) (bool, string) {    f.Lock()    defer f.Unlock()
    if _, exists := f.limits[userID]; !exists {        // 如果用户没有限流器，则创建一个新的。        f.limits[userID] = NewRateLimiterV2(capacity, refillRate, limit)    }
    // 使用用户的限流器检查请求。    return f.limits[userID].AllowWithFeedback()}

// FlexibleLimiter 结构体是一个灵活的限流器，允许在运行时动态调整限流参数。type FlexibleLimiter struct {    sync.Mutex // 使用sync.Mutex提供互斥锁功能，确保线程安全。    capacity  int  // 桶的容量，表示最多可以存储的令牌数。    refillRate float64 // 令牌的填充速率，表示每秒可以新增的令牌数。    limit      int  // 请求处理的阈值，用于确定是否限流。}
// SetParams 方法允许动态设置FlexibleLimiter的限流参数。// 这些参数包括桶的容量、填充速率和请求处理的阈值。func (f *FlexibleLimiter) SetParams(capacity int, refillRate float64, limit int) {    f.Lock() // 使用互斥锁进入临界区，防止并发访问导致的数据不一致。    defer f.Unlock() // 离开临界区前自动释放锁。
    // 更新FlexibleLimiter的参数。    f.capacity, f.refillRate, f.limit = capacity, refillRate, limit}
// Allow 方法根据FlexibleLimiter当前的参数决定是否允许新的请求。// 它首先基于当前参数创建一个新的RateLimiterV2实例，然后调用它的AllowWithFeedback方法。func (f *FlexibleLimiter) Allow() (bool, string) {    // 根据FlexibleLimiter当前的容量、填充速率和阈值创建一个新的RateLimiterV2实例。    rl := NewRateLimiterV2(f.capacity, f.refillRate, f.limit)    // 调用RateLimiterV2的AllowWithFeedback方法，获取是否允许请求的反馈。    // 这个方法返回一个布尔值表示是否允许请求，和一个字符串消息提供反馈信息。    return rl.AllowWithFeedback()}

// TransparentLimiter 结构体嵌入了RateLimiterV2，提供了额外的状态信息，// 包括当前剩余的令牌数，以增强限流机制的透明性。type TransparentLimiter struct {    *RateLimiterV2 // 嵌入RateLimiterV2，获得其所有功能。    currentTokens int // 存储当前桶中剩余的令牌数。}
// AllowWithStatus 方法允许请求并返回当前限流状态。// 它调用内嵌RateLimiterV2的AllowWithFeedback方法来决定是否允许请求，// 并获取反馈消息，同时返回当前剩余的令牌数。func (t *TransparentLimiter) AllowWithStatus() (bool, string, int) {    allowed, message := t.RateLimiterV2.AllowWithFeedback() // 调用内嵌限流器的允许逻辑。    return allowed, message, t.currentTokens // 返回是否允许、消息和当前令牌数。}
// MiddlewareWithTransparency 函数创建一个中间件，用于在HTTP响应中包含限流状态。// 这个中间件包装了下一个http.Handler，并在处理请求之前检查限流状态。func MiddlewareWithTransparency(next http.Handler) http.Handler {    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {        // 创建或使用全局的transparentLimiter实例来检查限流状态。        allowed, message, tokens := transparentLimiter.AllowWithStatus()
        // 如果请求被限流（不允许），则设置HTTP头部信息和状态码，并返回错误消息。        if !allowed {            w.Header().Set("X-RateLimit-Remaining", fmt.Sprintf("%d", tokens)) // 设置剩余令牌数的头部。            w.WriteHeader(http.StatusTooManyRequests)                             // 设置HTTP状态码为429。            fmt.Fprintln(w, message)                                            // 写入错误消息到响应体。            return                                                                // 中断请求处理。        }
        // 如果请求未被限流，继续执行后续的处理链。        next.ServeHTTP(w, r)    })}