软件系统限流的底层原理解析

// 这是一个简单的实现案例
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")
       }
    }
}

// 这是一个伪代码案例，演示实现逻辑
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/;
        }
    }
}

// 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()
    }
}

// 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)
    })
}