Redis之zset实现滑动窗口限流

import time
import redis
 
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
 
# 滑动窗口限流的实现
def is_action_allowed(key, max_count, duration):
    # 获取当前窗口的起始时间点
    window_start = int(time.time()) - (int(time.time()) % duration)
    
    # 计算窗口结束时间点
    window_end = window_start + duration
    
    # 计算当前窗口的key
    window_key = f"{key}:{window_start}-{window_end}"
    
    # 使用Redis的原子操作来限流
    with r.pipeline() as pipe:
        # 检查是否超出了限制
        pipe.exists(window_key)
        pipe.expire(window_key, duration + 1)  # 确保窗口有足够的生命周期
        # 增加计数
        pipe.incr(window_key)
        pipe.expire(window_key, duration + 1)  # 确保窗口有足够的生命周期
        # 检查当前计数是否超过限制
        pipe.pttl(window_key)
        
        # 执行管道中的命令
        _, _, current_count, ttl = pipe.execute()
        
        # 如果当前计数超过了限制，返回False
        if current_count and current_count > max_count:
            return False
        # 否则，返回True
        return True if ttl > 0 else False
 
# 使用示例
key = "rate_limit:action"
max_count = 10
duration = 60  # 窗口时长为60秒
 
# 模拟用户请求
for i in range(20):
    allowed = is_action_allowed(key, max_count, duration)
    print(f"Request {i}: {'Allowed' if allowed else 'Denied'}")
    time.sleep(3)  # 每3秒发送一个请求

这段代码使用Redis的原子操作来实现一个基于滑动窗口的限流器。它会检查是否超出了指定时间窗口内的请求限制，并相应地返回是否允许继续执行操作。