回答

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