API限流实战：三种主流算法实现

如果不做限流，一个写错死循环的客户端或者一次简单的DDoS就能把后端直接冲垮，云账单还能让你心惊肉跳。限流的核心就是控制单位时间内客户端能发起的请求数。

我整理了三种常见的实现方案，直接上代码。

一、固定窗口计数法 (Fixed Window)
最简单的逻辑：给每个用户开个计数器，时间窗到了就清零。

import time
from collections import defaultdict

class FixedWindowLimiter:
    def __init__(self, max_requests=10, window_seconds=60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)

    def allow_request(self, user_id):
        now = time.time()
        window_start = now - self.window_seconds
        # 剔除过期记录
        self.requests[user_id] = [t for t in self.requests[user_id] if t > window_start]
        if len(self.requests[user_id]) >= self.max_requests:
            return False
        self.requests[user_id].append(now)
        return True

• 优点： 实现极快。
• 缺点： 临界点容易翻倍。比如第59秒发10个，第60秒又发10个，瞬间吞吐量就翻倍了。

二、滑动窗口日志 (Sliding Window Log)
不搞固定时间段，而是记录每个请求的时间戳，动态计算窗口内的数量。

import time
from collections import defaultdict

class SlidingWindowLimiter:
    def __init__(self, max_requests=10, window_seconds=60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.logs = defaultdict(list)

    def allow_request(self, user_id):
        now = time.time()
        cutoff = now - self.window_seconds
        self.logs[user_id] = [t for t in self.logs[user_id] if t > cutoff]
        if len(self.logs[user_id]) >= self.max_requests:
            return False
        self.logs[user_id].append(now)
        return True

• 优点： 解决了临界点突刺问题。
• 缺点： 内存开销大，请求量一旦上百万，存储时间戳会非常吃资源。

三、令牌桶 (Token Bucket)
这是我最推荐的方案。桶里匀速放令牌，请求得拿令牌才能通过，允许一定的突发流量。

import time

class TokenBucket:
    def __init__(self, capacity=10, refill_rate=1):
        self.capacity = capacity
        self.refill_rate = refill_rate # 每秒恢复多少个
        self.tokens = capacity
        self.last_refill = time.time()

    def allow_request(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

• 优点： 兼顾了平滑限流和突发流量的处理，内存占用极低。
• 缺点： 逻辑比前两种稍微复杂一点。

避坑指南

在实际部署到生产环境时，有几个点必须注意：

• 内存泄漏： 如果用内存 Map 存储，记得清理过期用户数据，否则内存会一直涨。
• 分布式环境： 单机限流在多实例部署时失效，必须用 Redis 或 Memcached 做统一计数。
• 响应头规范： 记得在 HTTP Header 返回 X-RateLimit-Limit 和 X-RateLimit-Remaining，让调用方知道什么时候能重试。

如果是高并发的支付接口或实时服务，直接上令牌桶+Redis方案。