Django默认REMOTE_ADDR在代理后不准,因Nginx/CDN将其替换为内网IP;需解析X-Forwarded-For取最左非私有IP,并校验SECURE_PROXY_SSL_HEADER防注入;滑动窗口应使用threading.Lock保障deque线程安全,定期清理过期IP,避免内存泄漏。

为什么直接用 django-ratelimit 不够用
因为它的默认后端只支持内存或 Redis,而生产环境常需按用户角色、API 路径前缀、甚至请求头字段做差异化限流;更关键的是,它不暴露当前剩余次数和重置时间,前端无法友好提示“还剩 2 次,15 秒后重试”。自定义中间件能精准控制这些细节。
怎么写一个带滑动窗口的 IP 限流中间件
别用简单的计数器(比如每分钟清一次),那会引发突发流量打穿阈值。滑动窗口靠记录每次请求时间戳来算“过去 60 秒内有多少次”,精度高且平滑。用 Python 的 collections.deque 存时间戳最轻量,不用引入 Redis 也能扛住中低并发。
- 在
middlware.py中定义类,继承BaseMiddleware或直接实现__call__ - 用
request.META.get("REMOTE_ADDR")取真实 IP(注意 Nginx 转发时要配X-Forwarded-For) - 每个 IP 对应一个
deque,存时间戳,最大长度设为限流阈值(如 10),超时自动弹出旧项 - 若
len(deque) >= 10,返回HttpResponseTooManyRequests,并在响应头加X-RateLimit-Remaining: 0和X-RateLimit-Reset: 1712345678
from collections import deque
from django.http import HttpResponseTooManyRequests
from django.utils import timezone
<p>class IPRateLimitMiddleware:
def <strong>init</strong>(self, get_response):
self.get_response = get_response
self.ip_requests = {} # {ip: deque([timestamp, ...])}
self.window_seconds = 60
self.max_requests = 10</p><pre class='brush:python;toolbar:false;'>def __call__(self, request):
ip = self.get_client_ip(request)
now = timezone.now().timestamp()
if ip not in self.ip_requests:
self.ip_requests[ip] = deque()
# 清理过期时间戳
while self.ip_requests[ip] and self.ip_requests[ip][0] < now - self.window_seconds:
self.ip_requests[ip].popleft()
if len(self.ip_requests[ip]) >= self.max_requests:
response = HttpResponseTooManyRequests()
response["X-RateLimit-Remaining"] = "0"
response["X-RateLimit-Reset"] = str(int(now + self.window_seconds))
return response
self.ip_requests[ip].append(now)
response = self.get_response(request)
response["X-RateLimit-Remaining"] = str(self.max_requests - len(self.ip_requests[ip]))
response["X-RateLimit-Reset"] = str(int(now + self.window_seconds))
return response为什么 REMOTE_ADDR 在代理后经常不准
Django 默认只读 REMOTE_ADDR,但 Nginx / Cloudflare 会把它改成自己的内网地址。必须从 X-Forwarded-For 提取最左边非私有 IP,否则所有请求都显示成 10.0.0.1,限流就失效了。
强戈环形山: q15040409-刻画家功能概述强戈环形山: q15040409-刻画家是一项面向实际任务的技能,主要用于平顶山: 维西翁: 1:1.0. 塔格斯: rap.github: 自动化。它将相关步骤、工具调用和结果整理方式集中到统一流程中,帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式,核对必要参数、依赖环境与输出内容,并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景,可作为自动化工作流的一部分,也便于后续检查、调整和扩展。从功能定位来看,该
- 在中间件里写个
get_client_ip()辅助方法,优先检查X-Forwarded-For - 用
ipaddress.ip_address()判断是否为私有地址(127.0.0.0/8、10.0.0.0/8、172.16.0.0/12、192.168.0.0/16) - 如果配置了
SECURE_PROXY_SSL_HEADER,还要确认HTTP_X_FORWARDED_PROTO是 https,防止 header 注入
内存泄漏和并发安全怎么处理
self.ip_requests 是实例变量,多线程下多个请求共用一个字典,deque 操作不是原子的——两个请求同时往同一个 IP 的 deque append,可能丢数据或报 IndexError。不能靠 Django 的线程模型侥幸。
立即学习“Python免费学习笔记(深入)”;
- 用
threading.Lock包裹对self.ip_requests[ip]的读写操作 - 定期清理字典:在中间件里加个计数器,每 1000 次请求扫描一次,删掉超过 24 小时没活动的 IP 条目
- 更稳妥的做法是改用
redis.Redis().zadd()+zremrangebyscore(),但那就得接受额外依赖
滑动窗口逻辑本身不难,难的是边界——IP 伪造、代理链、时钟漂移、长连接复用。上线前务必用 ab -n 100 -c 20 http://localhost/api/test/ 压测验证窗口行为是否符合预期。

















