Flask 实战：开发轻量的接口限流服务

前言

在后端服务开发中，接口限流是保障服务稳定性必不可少的手段。不管是防止爬虫高频抓取数据，还是避免用户重复提交请求，都需要限流策略保护后端不被突发流量打垮。

很多同学一听到限流，第一反应就是 Redis + Lua。但在小型项目、内部工具原型阶段，我们完全可以先用内存滑动窗口实现基础限流逻辑，理解限流底层原理。等业务规模变大之后，再迁移到分布式Redis限流方案。本文从零搭建一个带限流功能的Flask服务，带你动手实现。

技术栈

• Python3：开发语言
• Flask：轻量Web框架，快速搭建HTTP接口
• time：内置时间模块，用于时间戳计算
• collections.deque：双端队列，用来维护滑动窗口内的请求记录

项目初始化

创建项目文件夹，安装依赖

pip install flask

新建 app.py，完整代码如下：

from flask import Flask, request, jsonify
from collections import deque
import time

app = Flask(__name__)

# 内存存储：key为客户端IP，value是双端队列，存放请求时间戳
request_records = {}
# 限流配置：窗口60秒，最多允许5次请求
WINDOW_SIZE = 60
MAX_REQUEST = 5

def rate_limit_middleware():
    """滑动窗口限流中间件"""
    client_ip = request.remote_addr
    now = time.time()
    # 不存在该IP记录则初始化队列
    if client_ip not in request_records:
        request_records[client_ip] = deque()
    queue = request_records[client_ip]

    # 清理窗口外过期的请求记录
    while queue and queue[0] < now - WINDOW_SIZE:
        queue.popleft()
    
    if len(queue) >= MAX_REQUEST:
        return jsonify({
            "code": 429,
            "msg": "请求过于频繁，请稍后再试",
            "data": None
        }), 429
    # 记录本次请求时间戳
    queue.append(now)
    return None

@app.before_request
def before_request():
    # 对所有接口执行限流校验
    res = rate_limit_middleware()
    if res:
        return res

@app.route("/api/hello", methods=["GET"])
def hello():
    return jsonify({
        "code": 200,
        "msg": "请求成功",
        "data": "Hello Flask Rate Limit"
    })

@app.route("/api/info", methods=["GET"])
def info():
    return jsonify({
        "code": 200,
        "msg": "请求成功",
        "data": "这是受限流保护的业务接口"
    })

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000, debug=False)

核心原理讲解

滑动窗口限流，是固定窗口的优化版本。固定窗口会出现窗口临界点流量突增的问题，滑动窗口把时间窗口切分成更小的粒度，持续剔除过期请求记录，统计当前窗口内的请求总数。

我们使用 deque 双端队列保存每个IP的请求时间戳：

1. 收到请求，拿到客户端IP与当前时间戳；
2. 循环弹出队列头部早于窗口起始时间的记录；
3. 判断队列剩余请求数量，如果超过阈值，直接返回429限流；
4. 未超限，则把当前请求时间戳加入队列，放行接口。

本实现是内存版，服务重启后所有限流记录丢失，只适合单机原型。

本地测试

运行服务：

python app.py

访问接口：[http://127.0.0.1:5000/api/hello](http://127.0.0.1:5000/api/hello)
连续快速刷新，前5次正常返回200，第6次及之后返回429提示请求频繁。等待60秒后计数器自动重置，又可以正常请求。

可以使用 curl 批量压测测试效果：

curl [http://127.0.0.1:5000/api/hello](http://127.0.0.1:5000/api/hello)

实战踩坑

1. 内存版本不支持多实例分布式部署
   如果服务多机负载均衡，内存数据不共享，单机限流会失效。分布式场景必须迁移Redis。
2. 识别客户端IP问题
   如果服务前面部署Nginx反向代理，request.remote_addr 获取到的是Nginx地址，需要配置代理头 X-Forwarded-For 获取真实用户IP。
3. 内存持续膨胀
   极端情况下大量不同IP访问，字典会无限增长。生产原型需要增加定时清理过期IP记录的逻辑，防止内存泄漏。

生产环境优化方向

1. 替换存储：使用 Redis + sorted set 实现分布式滑动窗口，支持多实例共享限流计数；
2. 灵活配置：针对不同接口配置不同限流阈值，比如登录接口限制更严格；
3. 限流key多样化：除IP外，可以使用用户ID作为限流key，适合登录态接口；
4. 增加日志：记录触发限流的请求，方便排查恶意爬虫；
5. 降级策略：限流时可以返回友好提示，或者直接开启接口降级。

总结

接口限流是后端稳定性防护体系里非常基础的一环。本文通过Flask+内存滑动窗口，用简短代码实现了可运行的限流原型。动手实践之后，就能理解滑动窗口的核心逻辑，后续再学习Redis分布式限流就会轻松很多。