用 Python + Redis 实现轻量级接口限流的入门实践

接口限流不是为了把用户挡在门外，而是为了在流量突增时保住核心服务。没有限流时，一个被反复刷新的列表接口、一个被脚本循环调用的短信接口，都可能把数据库和下游服务拖垮。限流的本质，是给每个调用方设定一个可预期的预算，超过预算就温和地拒绝，而不是等系统崩溃后再抢救。

先想清楚：限制谁、限制什么

常见的限流维度有三种。按用户限制，适合登录后的高频操作；按 IP 限制，适合未登录的公开接口；按接口限制，适合成本高、容易被滥用的关键路径。实际项目里常常组合使用，例如短信接口同时按手机号和 IP 限制，普通查询接口只按用户限制。先定好维度，再谈数值，否则很容易出现一个人刷爆、所有人受牵连的情况。

最简单的实现：固定窗口计数

固定窗口的思路很好理解：把时间切成一段一段，例如每 60 秒一个窗口，同一个键在窗口内计数，超过上限就拒绝。Redis 很适合做这件事，因为 INCR 是原子的，多个进程同时调用也不会把计数算错。第一次写入时设置过期时间，窗口结束计数自动清零，不需要定时任务清理。

import redis
import time

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

def allow(key, limit=100, window=60):
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)
    return count <= limit

调用时把键拼成 limit:用户ID:接口路径 这种形式，既能区分用户，也能区分接口。如果返回 False，就返回 429 状态码，并告诉调用方稍后再试。不要直接返回 500，因为被限流不是服务出错，而是调用方需要放慢节奏。

用装饰器把限流收敛到一处

如果每个接口都手写计数逻辑，代码很快就会散落各处。更好的做法是写一个装饰器，把限流逻辑集中管理。装饰器里先拼键、再判断，超限直接返回统一的错误结构，业务函数只关心正常流程。这样做的好处是规则可改：把上限从 100 改成 200，只需要改一处配置，而不是翻遍所有接口。

固定窗口的两个坑

第一个坑是窗口边界突刺。假设上限是每分钟 100 次，有人在第 59 秒调用 100 次，第 61 秒又调用 100 次，两分钟内实际是 200 次，却都通过了检查。对一般业务接口可以接受，但对短信、支付这类敏感接口就不够严谨。更精细的做法是用滑动窗口或令牌桶，代价是实现更复杂、Redis 操作更多，需要按业务风险取舍。

第二个坑是把所有流量都当成同一种流量。健康检查、内部调用、搜索引擎抓取如果和普通用户走同一条限流规则，可能会被误伤。常见做法是白名单内部调用、给关键接口单独设置更宽松的上限，并对限流结果做监控：被拒绝次数突然暴涨，往往意味着有异常流量或者规则配错了，而不是用户突然变多了。

上线前要验证的三件事

第一，验证并发正确性。用多线程同时请求，看计数是否准确，有没有因为非原子操作导致超发。第二，验证过期时间。键的过期时间应该和窗口一致，否则窗口结束后计数不清零，会把正常用户一直挡在外面。第三，验证失败兜底。Redis 临时不可用时，是直接放行还是直接拒绝，需要提前定好。放行会让限流失效，拒绝可能影响正常业务，多数内部服务会选择记录日志并放行，同时触发告警。

结语

轻量级限流不需要一开始就引入复杂的中间件。先用 Redis 固定窗口把核心接口保护起来，跑通监控和告警，再根据真实流量决定是否升级到滑动窗口或令牌桶。把限流做成可配置、可观测、可回滚的一层防护，系统在流量高峰时才能从容降级，而不是被动崩溃。