django中全局限流需用redis实现原子计数,因django-ratelimit默认基于本地缓存,多进程下计数不共享;应使用incr+expire(带nx)组合、pipeline保证原子性,并设计含用户、路径、时间维度的key。

为什么直接用 django-ratelimit 可能不满足全局限流需求
因为 django-ratelimit 默认基于内存(cache 后端),若部署多进程或多机器,各进程缓存独立,计数不共享,实际变成“单进程限流”。真正全局限流必须依赖外部共享存储,比如 Redis —— 这不是配置开关能解决的,得自己控制计数逻辑和原子操作。
如何用 Redis + 中间件实现原子性请求计数
核心是:每次请求进来,在 Redis 里对一个带 TTL 的 key 做原子自增,并检查是否超限。不能先 GET 再 INCR,否则并发时会漏判;必须用 INCR + EXPIRE 组合,且 EXPIRE 要用 nx 参数避免重复设置过期时间。
- 用
redis-py的pipeline包裹incr和条件expire,保证原子性 - 限流 key 要包含足够维度,例如:
f"rate:{user_id}:{path}:{minute_timestamp}",避免不同用户或路径互相干扰 - 不要在中间件里直接调用
time.time()算分钟戳,用datetime.now().replace(second=0, microsecond=0)更可靠 - 如果没登录用户,可用
request.META.get("REMOTE_ADDR")或带签名的X-Forwarded-For(注意代理可信度)
中间件中拦截并返回 429 的正确姿势
Django 中间件的 process_request 是最早可介入的钩子,但此时响应还没生成,适合做拦截;别用 process_view,它晚于 URL 解析,可能已触发视图逻辑或 DB 查询。
- 判断超限时,直接返回
HttpResponse("Too Many Requests", status=429),不要 raise 异常 - 务必设置响应头:
response["X-RateLimit-Limit"] = "100"、response["X-RateLimit-Remaining"] = str(remaining)、response["Retry-After"] = "60" - 如果用 class-based middleware,记得继承
BaseMiddleware或实现__call__,Django 4.1+ 推荐函数式写法更轻量 - 避免在限流逻辑里查数据库或调远程服务,否则限流本身成性能瓶颈
Redis 连接与异常降级怎么处理
Redis 不可用时,全局限流失效比误拦更危险——宁可放行也不能卡住所有请求。所以中间件里必须有兜底逻辑。
- 用
try/except redis.ConnectionError, redis.TimeoutError:捕获连接问题,记录 warning 日志后跳过限流 - 不要用
redis.from_url(..., socket_connect_timeout=0.1)这种硬超时,容易因瞬时抖动误降级;建议设为0.3并配合重试次数 1 - 本地开发用
LocMemCache模拟不可行:它不支持INCR,必须真实 Redis 或redislite测试 - 生产环境 Redis 建议用连接池(
ConnectionPool),避免每次新建连接,尤其高 QPS 场景
rate_limit_exceeded 指标和 Redis latency 曲线。Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











