Python 中有几个实用的限流库,以下是一些推荐选项及其特点:
throttled-py
这是一个轻量且灵活的限流库,支持多种限流算法(如固定窗口、滑动窗口、令牌桶、漏桶和 GCRA),适配 Redis 存储后端,同时支持同步和异步场景。它还提供线程安全的存储后端(如内存和 Redis),支持配置限流策略及等待重试模式。
安装
pip install throttled-py
使用示例(简单版)
以下是一个使用 内存存储 + 令牌桶算法 的基本限流示例:示例 1:限制每秒最多处理 3 次请求
from throttled_py.memory import MemoryStore from throttled_py.limiter import RateLimiter from throttled_py.strategy.token_bucket import TokenBucket # 存储方式:内存 store = MemoryStore() # 创建限流策略:每秒最多 3 次 strategy = TokenBucket(rate=3, capacity=3) # 创建限流器 limiter = RateLimiter(store=store, strategy=strategy) # 模拟请求 for i in range(5): if limiter.allow("user_123"): # "user_123" 是标识符(如用户ID、IP) print(f"请求 {i+1}: 允许") else: print(f"请求 {i+1}: 被拒绝")
输出可能如下:请求 1: 允许请求 2: 允许请求 3: 允许请求 4: 被拒绝请求 5: 被拒绝
分布式限流(Redis 支持)如果你是分布式服务(比如多个节点部署),可以使用 Redis 后端:
from redis import asyncio as aioredis from throttled_py.redis import RedisStore from throttled_py.limiter import RateLimiter from throttled_py.strategy.token_bucket import TokenBucket # 异步连接 Redis redis_client = aioredis.from_url("redis://localhost") # Redis 存储 store = RedisStore(redis=redis_client, key_prefix="rate_limit:") # 限流策略:每秒最多 5 次 strategy = TokenBucket(rate=5, capacity=5) # 创建限流器 limiter = RateLimiter(store=store, strategy=strategy) # 在 async 函数中使用 import asyncio async def test(): for i in range(7): allowed = await limiter.are_allow("user_456") print(f"请求 {i+1}: {'允许' if allowed else '被拒绝'}") asyncio.run(test())
ratelimit
基于令牌桶算法实现的速率限制库,通过装饰器简化对函数或方法的速率限制配置,适合需要简洁方案的场景。ratelimiter
专注于控制函数或方法的调用频率,防止系统过载或滥用,适用于需要严格限制调用次数的场景。
分布式限流(GCRA 实现)
对于分布式系统,可通过基于 GCRA(通用信元速率算法)的限流方案实现,结合 Redis 和 Lua 脚本(或翻译为 Python 实现)完成高效限流。
适用场景建议 :
单机环境可选择 ratelimit 或 ratelimiter;
复杂场景(如多算法支持、分布式限流)推荐 throttled-py 或基于 GCRA 的自定义实现网友回复