1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import time

class FixedWindowRateLimiter:
    """固定窗口计数器：实现最简单，但有边界双倍突发的风险"""
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window = window_seconds
        self.count = 0
        self.window_start = int(time.time())

    def allow(self) -> bool:
        now = int(time.time())
        if now - self.window_start >= self.window:
            # 进入新窗口
            self.window_start = now
            self.count = 0
        if self.count < self.limit:
            self.count += 1
            return True
        return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import time
from collections import deque

class SlidingWindowRateLimiter:
    """滑动窗口计数器：精度高，能避免固定窗口的边界双倍突发"""
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window = window_seconds
        # 存 (时间戳, )，每次清理掉超过 window 的旧时间戳
        self.timestamps: deque[float] = deque()

    def allow(self) -> bool:
        now = time.time()
        # 弹出过期的时间戳
        while self.timestamps and self.timestamps[0] <= now - self.window:
            self.timestamps.popleft()
        if len(self.timestamps) < self.limit:
            self.timestamps.append(now)
            return True
        return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import time
from collections import deque

class LeakyBucketRateLimiter:
    """漏桶算法：强制平滑输出，适合下游处理能力固定的场景"""
    def __init__(self, capacity: int, leak_rate_per_sec: float):
        """
        :param capacity: 桶容量
        :param leak_rate_per_sec: 每秒漏出的水滴数
        """
        self.capacity = capacity
        self.leak_rate = leak_rate_per_sec
        self.queue: deque[float] = deque()
        self.last_leak_time = time.time()

    def _leak(self):
        """把超过当前时间该漏出的水滴都漏掉"""
        now = time.time()
        elapsed = now - self.last_leak_time
        leaked = int(elapsed * self.leak_rate)
        if leaked > 0:
            for _ in range(min(leaked, len(self.queue))):
                self.queue.popleft()
            self.last_leak_time = now

    def allow(self) -> bool:
        self._leak()
        if len(self.queue) < self.capacity:
            self.queue.append(time.time())
            return True
        return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import time
import threading

class TokenBucketRateLimiter:
    """令牌桶算法：兼顾平滑限流 + 允许突发，是工业界最常用的限流算法"""
    def __init__(self, capacity: int, refill_rate_per_sec: float):
        """
        :param capacity: 桶容量（即最大突发量）
        :param refill_rate_per_sec: 每秒生成的令牌数（平均限流速率）
        """
        self.capacity = capacity
        self.refill_rate = refill_rate_per_sec
        self.tokens = float(capacity)  # 初始满桶
        self.last_refill_time = time.time()
        self._lock = threading.Lock()

    def allow(self, cost: float = 1.0) -> bool:
        with self._lock:
            now = time.time()
            # 补充令牌：自上次以来应该生成的令牌数
            elapsed = now - self.last_refill_time
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            self.last_refill_time = now
            # 取令牌
            if self.tokens >= cost:
                self.tokens -= cost
                return True
            return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// 创建一个每秒生成 5 个令牌的限流器（即平均 200ms 一个）
RateLimiter limiter = RateLimiter.create(5.0);

if (limiter.tryAcquire()) {
    // 拿到令牌，处理请求
    handleRequest();
} else {
    // 没拿到令牌，直接拒绝
    return Response.status(429).build();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 定义一个名为 my_limit 的限流区：
#   10m 共享内存大小
#   rate=10r/s 平均 10 个请求/秒
#   burst=20 桶容量 20（允许最大 20 个突发）
#   nodelay 突发请求不延迟直接处理
limit_req_zone $binary_remote_addr zone=my_limit:10m rate=10r/s;

server {
    location /api/ {
        limit_req zone=my_limit burst=20 nodelay;
        proxy_pass http://backend;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

-- KEYS[1]: 令牌桶的 key
-- ARGV[1]: 桶容量
-- ARGV[2]: 填充速率（每秒）
-- ARGV[3]: 当前时间戳（秒）
-- ARGV[4]: 请求消耗的令牌数

local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])

local data = redis.call('HMGET', KEYS[1], 'tokens', 'last_refill')
local tokens = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now

-- 补充令牌
local elapsed = math.max(0, now - last_refill)
tokens = math.min(capacity, tokens + elapsed * rate)

if tokens >= cost then
    tokens = tokens - cost
    redis.call('HMSET', KEYS[1], 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', KEYS[1], 60)  -- 60 秒没访问就清掉
    return 1
else
    redis.call('HMSET', KEYS[1], 'tokens', tokens, 'last_refill', now)
    return 0
end