+
34
-

回答

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 的自定义实现

网友回复

我知道答案,我要回答