Python中如何实现多线程和多进程环境下的Redis限流器？

核心思路：
在Python的多线程/多进程环境下实现Redis限流器，关键在于利用Redis的原子操作（如INCR + EXPIRE）或更精确的令牌桶/漏桶算法，并通过分布式锁或Redis的原子性保证并发安全。以下是两种常见方案的代码示例：

方案1：固定窗口计数器（简单场景）

适用于对精度要求不高的限流，使用INCR和EXPIRE实现：

import redis
import time

class RedisRateLimiter:
    def __init__(self, redis_client, key, max_requests, window_seconds):
        self.redis = redis_client
        self.key = key
        self.max_requests = max_requests
        self.window = window_seconds

    def is_allowed(self):
        # 原子操作：计数+过期时间设置
        current = self.redis.incr(self.key)
        if current == 1:
            self.redis.expire(self.key, self.window)
        return current <= self.max_requests

# 使用示例（多线程/进程共享同一Redis客户端）
redis_client = redis.Redis(host='localhost', port=6379)
limiter = RedisRateLimiter(redis_client, "api_limit:user1", 100, 60)

# 在多个线程/进程中调用
if limiter.is_allowed():
    print("请求通过")
else:
    print("请求被限流")

方案2：令牌桶算法（平滑限流）

使用Redis的LIST存储令牌，通过BRPOP实现阻塞获取令牌：

import redis
import threading

class TokenBucketLimiter:
    def __init__(self, redis_client, key, refill_rate, capacity):
        self.redis = redis_client
        self.key = key
        self.refill_rate = refill_rate  # 令牌/秒
        self.capacity = capacity
        self._init_bucket()

    def _init_bucket(self):
        # 初始化桶满令牌
        if not self.redis.exists(self.key):
            self.redis.lpush(self.key, *['token'] * self.capacity)

    def acquire_token(self, timeout=1):
        # 阻塞获取令牌（BRPOP是原子操作）
        token = self.redis.brpop(self.key, timeout=timeout)
        if token:
            # 异步补充令牌（可通过后台线程或Redis定时任务实现）
            self._refill_bucket()
            return True
        return False

    def _refill_bucket(self):
        # 简化的补充逻辑：实际需根据时间间隔计算补充数量
        current_size = self.redis.llen(self.key)
        if current_size < self.capacity:
            self.redis.lpush(self.key, 'token')

# 多线程测试示例
def worker(limiter, id):
    if limiter.acquire_token():
        print(f"线程{id}获取令牌成功")
    else:
        print(f"线程{id}被限流")

redis_client = redis.Redis()
limiter = TokenBucketLimiter(redis_client, "token_bucket:api", refill_rate=10, capacity=50)
threads = [threading.Thread(target=worker, args=(limiter, i)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

关键点说明

1. 原子性保障：所有限流操作必须使用Redis的原子命令（如INCR、BRPOP、Lua脚本），避免并发竞争。
2. 多进程支持：确保不同进程连接同一Redis实例，方案与多线程完全相同。
3. 时间窗口同步：依赖Redis服务器时间，避免客户端时间不同步问题。
4. 高性能建议：使用连接池管理Redis连接，Lua脚本合并多个操作减少网络往返。

总结：根据业务精度需求选择固定窗口或令牌桶方案，核心是依赖Redis的原子操作。