固定窗口限流算法
我们都知道 Redis 中间件 可以做很多事情，最常用的可以作为缓存数据库， 用于缓存热点数据，做排行榜，以及做消息队列等等。
本文想说一下 Redis 本身也可以用来做接口限流，比如对于某些接口 由于并发的限制，需要一个限制比如60s 允许 60个请求，后面的请求 我们来返回特定的状态码，来保证接口服务 提供正常的服务。
算法核心思想
Redis固定窗口限流思想是一种简单的流量控制策略，主要用于限制在特定时间间隔内允许的请求或事件数量。它的核心理念是将时间线划分成一系列固定长度的时间段（窗口），每个窗口内设定一个请求的最大数量限制。当请求到来时，会检查当前窗口是否已达到或超过预设的请求上限，以此决定是否继续处理请求。
看看 下面 的一张图示意图， 这里假如我们限制在 60s 内 允许有6个请求，可以正常请求API， 那么我们可以把时间窗口 划分成如下， 60s 为一个window_seconds ,在这个窗口范围内的 允许请求的个数进行统计，如果没有超过限制就放行通过。
如何统计请求的数量，我们可以使用Redis 提供的 incr 命令，如果key 当前不存在 则设置这个key ,并且返回1，如果有这个key 是数值类型就会 自增1, 如果有这个key 但是不是数值类型，会抛出异常。
写了一个简单的装饰器来说明一下 如何限流的，这里只是演示一下这个情况。
import time
import redis
from functools import wraps
# Redis连接配置
redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0, password="your_password")
def rate_limit(max_requests: int, window_seconds: int):
"""
固定窗口限流装饰器。
:param max_requests: 允许的最大请求数。
:param window_seconds: 时间窗口长度（秒）。
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
api_key = func.__name__ # 假设使用函数名作为API标识符，可根据实际情况调整
current_time = int(time.time())
window_start = current_time - (current_time % window_seconds)
key = f"rate_limit:{api_key}:{window_start}"
request_count = redis_client.incr(key)
if request_count == 1:
redis_client.expireat(key, window_start + window_seconds)
if request_count > max_requests:
return "Too Many Requests", 429 # HTTP状态码429表示“太多请求”
else:
return func(*args, **kwargs)
return wrapper
return decorator
@rate_limit(max_requests=5, window_seconds=60)
def my_api_function():
"""示例API函数"""
return "API call successful."
if __name__ == "__main__":
# 测试API函数
for _ in range(20):
response = my_api_function()
print(response)
time.sleep(0.1) # 模拟请求间隔
# print(redis_client.ping())
pass
在限流策略中，window_start 变量代表了当前时间窗口的起始时间点。这个概念对于理解固定窗口限流策略至关重要。
固定窗口限流是将时间划分为一系列固定长度的窗口，每个窗口内允许的请求次数是固定的。为了实现这一点，我们需要知道当前请求落在哪个时间窗口内。这就需要计算出窗口的起始时间点，即window_start。
具体到代码实现中，window_start 是通过以下方式计算的：
current_time = int(time.time()) # 获取当前时间的Unix时间戳（秒）
window_start = current_time - (current_time % window_seconds)
这里的计算逻辑是：
time.time()返回当前时间的Unix时间戳，单位是秒。(current_time % window_seconds)计算出current_time在window_seconds时间段内的偏移量。例如，如果窗口长度是10秒，而当前时间是32秒，则这一计算结果为2秒，表示当前时间处于第3个10秒窗口内。current_time - (current_time % window_seconds)则是去除这个偏移量，得到的是最接近当前时间但刚好位于窗口起始位置的时间戳。这样就确定了当前请求所属的时间窗口的开始时间。
计算window_start的示例
假设当前时间戳current_time为14403秒（为了简化计算，这里假设秒是时间单位）。
如果我们的窗口大小window_seconds是10秒，那么我们想要找到当前时间属于哪个10秒窗口的起点，也就是window_start。
步骤解析
- 计算当前时间戳：
current_time = 14403秒 - 计算当前时间在窗口内的偏移量：
(current_time % window_seconds)，这里的%是取模运算，用来找出current_time除以window_seconds的余数。对于14403秒，除以10秒的余数是3秒，意味着当前时间位于某个10秒窗口的第3秒。 - 计算窗口起始时间：
window_start = current_time - (current_time % window_seconds)。将上面的计算代入，得到window_start = 14403 - 3 = 14400秒。这意味着当前时间窗口是从14400秒开始，持续到14409秒结束。
我们简单写一个应用 来测试一下，比如这里限制 60秒5 个请求
# -*- coding: utf-8 -*-
"""
@Time : 2024/5/1 19:35
@Author : Frank
@File : app.py
"""
from flask import Flask
from flask import jsonify
from util.rate_limit import rate_limit
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def index():
return "hello index"
@app.route('/info', methods=['GET', 'POST'])
@rate_limit(max_requests=5, window_seconds=60)
def info():
return jsonify({
'name': 'frank',
'gender': 'male',
'age': 18
})
def run(app):
app.run(host='0.0.0.0', port=5000)
if __name__ == '__main__':
run(app=app)
固定窗口限流的缺点
主要缺点包括以下几点：
- 临界点问题（毛刺现象）：在窗口切换的瞬间，所有限流指标会立即重置，这可能导致大量请求恰好在新窗口开始时涌入，造成瞬时流量尖峰，即所谓的“毛刺现象”。例如，如果限流速率是每秒10个请求，那么每秒的第一秒可能接收到远超10个请求。
- 无法平滑限流：由于固定窗口是以固定时间长度为单位统计请求，即使在窗口内早期已经达到限流阈值，后续进入的请求也必须等待整个窗口结束后才能被处理，这导致限流效果不够平滑，不能很好地应对短时间内的突发流量变化。
- 资源分配不均：在高并发场景下，特别是在窗口边界附近，可能会出现部分时间段内的请求被快速处理，而另一部分时间的请求则因达到限流阈值而被拒绝，即使总体并未超出系统的处理能力，造成资源分配不均。
- 不适合应对周期性波动流量：对于有明显周期性波动的流量模式，固定窗口算法可能在流量低谷时浪费配额，在流量高峰时又过于严格，不能灵活适应流量的变化。
固定窗口限流存在 以上的问题，对于高并发的系统来说， 如果每秒有几百个请求，确实也不好限制，无法实现平滑的限流 ，对于短时间的突发流量 也是不太友好. 对于不是特别高并发的系统来说，基本上可以使用了。
参考文档
https://blog.chenxiaosheng.com/posts/2022-03-28/rate-limit_fixed-window_sliding-window
https://redis.io/docs/latest/commands/incrby/
https://redis.io/docs/latest/commands/incr/
