QPS < 60

单机版

滑动窗口法

思路：为每个用户维护一个请求时间戳队列，每次请求时，从该用户的请求时间戳队列移出时间窗口以外的请求时间戳，然后统计队列剩余个数，如果小于qps，则允许请求并且添加时间戳到队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from time import time

class UserRateLimiter:
    def __init__(self, user_id, max_qps=60, window_size=1):
        self.user_id = user_id
        self.max_qps = max_qps          # 每秒最大请求数
        self.window_size = window_size  # 时间窗口（秒）
        self.request_timestamps = []    # 用户请求时间戳队列
    
    def allow_request(self):
	    # 返回是否允许请求
        current_time = time()
        
        # 移除时间窗口外的旧请求（滑动窗口）
        while self.request_timestamps and \
              self.request_timestamps[0] <= current_time - self.window_size:
            self.request_timestamps.pop(0)
        
        if len(self.request_timestamps) < self.max_qps:
            self.request_timestamps.append(current_time)
            return True  # 允许请求
        else:
            return False  # 拒绝请求

  • 数据结构：用队列存储请求时间戳，实现滑动窗口。
  • 时间复杂度：O(n)（可通过二分查找优化到 O(log n)）。
  • 优点：精确控制每秒请求数，无毛刺现象。

分布式版：

令牌桶+漏桶算法：

令牌桶：每个用户维护一个令牌桶，请求需要消耗令牌才能被处理。每次请求时，先计算需要增加的令牌，等于上次请求时间到现在的时间（秒）*qps,但不超过qps=60,然后查看令牌桶有无令牌，没有则拒绝，有则减少一个令牌

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from time import time

class TokenBucket:
    def __init__(self, capacity=60, fill_rate=60):
        self.capacity = capacity    # 桶容量
        self.tokens = capacity      # 当前令牌数
        self.fill_rate = fill_rate  # 令牌生成速率（个/秒）
        self.last_time = time()     # 上次更新时间

    def allow_request(self):
        current_time = time()
        elapsed = current_time - self.last_time
        
        # 计算新增令牌
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.fill_rate
        )
        self.last_time = current_time

        if self.tokens >= 1:
            self.tokens -= 1
            return True  # 允许请求
        return False     # 拒绝请求

令牌桶 vs 漏桶如何选择？

  • 选令牌桶：需要容忍突发（如用户快速连续点击）。
  • 选漏桶：严格要求匀速（如音频流传输）。

生产实践：

  1. 多级限流：用户级QPS限制+ 全局API网关限流。
  2. 黑名单机制
  3. 软限流：延迟返回