我要评论1. api速率限制的基本概念
api速率限制（rate limiting）是控制用户访问api的请求速率的一种机制，防止系统被过多请求淹没。通过对用户的请求进行计数与限制，api能够在高并发情况下维持性能与稳定性。
速率限制的主要目的是：
- 防止过多请求导致服务器负载过高。
- 限制恶意或非正常行为。
- 提高api的可用性，确保公平分配资源。
常见的速率限制算法
1.漏桶算法（leaky bucket algorithm）
漏桶算法在一定时间窗口内以固定速率处理请求，若请求速率超过预定限度，则会被丢弃或延迟。
2.令牌桶算法（token bucket algorithm）
令牌桶算法是一种灵活的速率控制机制，适合处理突发流量。每个请求都需要获取一个令牌，如果令牌桶为空，则请求被丢弃。
3.固定窗口计数法（fixed window counter）
在固定时间窗口内，计数器记录请求的次数，一旦请求超出限制，后续请求将被拒绝。
4.滑动窗口计数法（sliding window counter）
滑动窗口比固定窗口更加精细，每个请求都在一个滑动的时间窗口内进行计数，能平滑流量控制。
通过这些算法，api能够控制不同用户或客户端在指定时间内发起的请求数量，确保系统的平稳运行。
2. redis实现分布式速率限制
redis是一个高性能的键值数据库，广泛用于缓存、消息队列和分布式速率限制等场景。在分布式系统中，redis提供了高效的数据存储和共享机制，可以帮助不同服务器实例共享请求计数信息，从而实现跨服务器的速率限制。
redis的实现思路
我们使用redis的setex命令设置一个键值对，其中键为用户标识（例如ip或用户id），值为请求计数。每次用户发起请求时，我们先检查该键是否存在。如果存在，检查其值是否超过限额；如果不存在，设置新的键并开始计数。通过设置键的过期时间，可以实现速率限制。
示例代码
from fastapi import fastapi, request, httpexception
import redis
import time
app = fastapi()
# 连接redis服务器
r = redis.redis(host='localhost', port=6379, db=0)
# 限制参数
limit = 100 # 每分钟100次请求
time_window = 60 # 1分钟
@app.middleware("http")
async def rate_limit(request: request, call_next):
ip_address = request.client.host
current_time = int(time.time())
# 构造redis的键
redis_key = f"rate_limit:{ip_address}:{current_time // time_window}"
# 使用redis的incr命令增加计数
request_count = r.incr(redis_key)
if request_count == 1:
# 设置过期时间为60秒（时间窗口大小）
r.expire(redis_key, time_window)
if request_count > limit:
raise httpexception(status_code=429, detail="too many requests")
response = await call_next(request)
return response
代码解析
- r.incr(redis_key)：redis的incr命令可以原子性地递增键的值。如果键不存在，它会先创建键并设置初值为1。
- r.expire(redis_key, time_window)：设置键的过期时间，使得计数在每个时间窗口内自动重置。
- 429 too many requests：当请求次数超过限制时，返回429状态码表示超出请求频率限制。
这种方式可以有效防止单个ip地址在短时间内发送过多请求，保障api的可用性与性能。
3. 防止ddos攻击的常见策略
ddos（distributed denial of service）攻击通过大量恶意请求淹没目标服务器，导致系统不可用。为了防止这种攻击，除了传统的防火墙和负载均衡策略外，我们还需要在api层面实现防护。
常见的防御策略
1.ip黑名单/白名单
基于ip的访问控制可以有效阻止已知攻击源的流量。通过将恶意ip加入黑名单，可以防止这些ip的请求进入系统。
2.请求速率限制
利用速率限制算法（如漏桶或令牌桶），控制请求频率，避免单个来源发送过多请求。
3.行为分析与智能防护
通过分析请求的行为模式，识别并阻止异常流量。例如，检测异常的请求头、请求频率、请求路径等。
4.验证码与身份验证
在用户请求的关键环节，如登录、注册、支付等，加入验证码或二次身份验证，防止恶意机器人自动化攻击。
示例代码：基于ip的速率限制和验证码
from fastapi import fastapi, httpexception, request
from fastapi.responses import jsonresponse
from pydantic import basemodel
import redis
import time
import random
app = fastapi()
r = redis.redis(host='localhost', port=6379, db=0)
limit = 100
time_window = 60
captcha_threshold = 10
@app.post("/login")
async def login(request: request, user: basemodel):
ip_address = request.client.host
current_time = int(time.time())
redis_key = f"rate_limit:{ip_address}:{current_time // time_window}"
request_count = r.incr(redis_key)
if request_count == 1:
r.expire(redis_key, time_window)
if request_count > limit:
# 启动验证码机制
captcha = random.randint(1000, 9999)
return jsonresponse(content={"captcha_required": true, "captcha": captcha}, status_code=400)
return {"message": "login successful"}
代码解析
当请求频率超过限制时，返回一个验证码，用户需要通过验证码验证来继续操作。
这种方式有效阻止了自动化攻击，减少了恶意请求的成功率。
4. 基于ip或用户身份的访问频率控制
除了全局的速率限制外，还可以根据ip地址或用户身份来单独限制访问频率。通过这种方法，可以更精细化地控制api的访问权限，避免某个特定用户或ip占用过多资源。
示例代码：基于用户身份的访问频率控制
from fastapi import depends, httpexception, request
from pydantic import basemodel
@app.get("/user_dashboard")
async def user_dashboard(user_id: str, request: request):
user_limit_key = f"user:{user_id}:rate_limit"
ip_limit_key = f"ip:{request.client.host}:rate_limit"
# 用户访问频率限制
user_request_count = r.incr(user_limit_key)
if user_request_count == 1:
r.expire(user_limit_key, time_window)
if user_request_count > limit:
raise httpexception(status_code=429, detail="user request limit exceeded")
# ip访问频率限制
ip_request_count = r.incr(ip_limit_key)
if ip_request_count == 1:
r.expire(ip_limit_key, time_window)
if ip_request_count > limit:
raise httpexception(status_code=429, detail="ip request limit exceeded")
return {"message": "welcome to the user dashboard"}
代码解析
user_id：每个用户有独立的请求计数，防止某个用户滥用api。
request.client.host：ip地址的请求计数，防止同一个ip地址滥用api。
根据用户和ip的访问频率分别设置限制，提高了控制精度。
到此这篇关于使用redis实现请求限制与速率限制的文章就介绍到这了,更多相关redis请求限制与速率限制内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论