登录
首页 >  文章 >  python教程

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

时间:2026-05-11 19:18:41 180浏览 收藏

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

怎么在Python Django中利用Middleware限制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: 0X-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,限流就失效了。

  • 在中间件里写个 get_client_ip() 辅助方法,优先检查 X-Forwarded-For
  • ipaddress.ip_address() 判断是否为私有地址(127.0.0.0/810.0.0.0/8172.16.0.0/12192.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学习网公众号,一起学习编程~

资料下载
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>