Python Django限制IP访问频率的方法

本文深入解析了在Django生产环境中精准实现IP访问频率限制的关键技术难点与实战方案：针对代理（如Nginx/CDN）导致REMOTE_ADDR失真问题，强调必须安全解析X-Forwarded-For并校验私有IP与SSL头以防注入；摒弃简单计数器，采用线程安全的滑动窗口算法——基于deque动态维护60秒内请求时间戳，并通过threading.Lock和定期清理机制规避并发冲突与内存泄漏；同时指出django-ratelimit等第三方库在差异化限流（按角色、路径前缀、请求头）及前端友好反馈（实时返回剩余次数与重置时间）上的不足，手把手演示如何编写高可控、可扩展、可压测的自定义中间件，真正满足复杂业务场景下的精细化、高可靠限流需求。

为什么直接用 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
class IPRateLimitMiddleware:
def init(self, get_response):
self.get_response = get_response
self.ip_requests = {}  # {ip: deque([timestamp, ...])}
self.window_seconds = 60
self.max_requests = 10
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，限流就失效了。

• 在中间件里写个 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 的线程模型侥幸。

• 用 threading.Lock 包裹对 self.ip_requests[ip] 的读写操作
• 定期清理字典：在中间件里加个计数器，每 1000 次请求扫描一次，删掉超过 24 小时没活动的 IP 条目
• 更稳妥的做法是改用 redis.Redis().zadd() + zremrangebyscore()，但那就得接受额外依赖

滑动窗口逻辑本身不难，难的是边界——IP 伪造、代理链、时钟漂移、长连接复用。上线前务必用 ab -n 100 -c 20 http://localhost/api/test/ 压测验证窗口行为是否符合预期。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~