Redis实现限流的常见策略包括：
- 固定窗口限流器（Fixed Window Rate Limiter）
- 滑动窗口限流器（Sliding Window Rate Limiter）
- 漏桶算法（Leaky Bucket Algorithm）
- 令牌桶算法（Token Bucket Algorithm）
以下是使用Redis实现固定窗口限流器和滑动窗口限流器的示例代码：
固定窗口限流器：
import redis
import time
def is_rate_limited(redis_conn, user_id, max_requests, window_size):
key = f'rate_limit:{user_id}:{window_size}'
requests = redis_conn.incr(key)
if requests == 1:
redis_conn.expire(key, window_size)
if requests > max_requests:
return True
return False
r = redis.Redis()
user_id = 'user123'
max_requests = 10
window_size = 60 # 时间窗口为60秒
if is_rate_limited(r, user_id, max_requests, window_size):
print("You are being rate limited.")
else:
print("Request is allowed.")滑动窗口限流器：
import redis
import time
def is_rate_limited(redis_conn, user_id, max_requests, window_size):
key = f'rate_limit:{user_id}:{window_size}'
timestamp = str(time.time())
requests = redis_conn.incr(key)
if requests == 1:
redis_conn.expire(key, window_size)
else:
current_window = int(timestamp) // window_size * window_size
first_request_timestamp = redis_conn.get(key)
if first_request_timestamp:
if current_window > int(first_request_timestamp):
redis_conn.set(key, timestamp)
redis_conn.set(f'{key}:count', 1)
count = redis_conn.get(f'{key}:count')
if count and int(count) > max_requests:
return True
return False
r = redis.Redis()
user_id = 'user123'
max_requests = 10
window_size = 60 # 时间窗口为60秒
if is_rate_limited(r, user_id, max_requests, window_size):
print("You are being rate limited.")
else:
print("Request is allowed.")在这两个示例中，我们使用Redis的INCR命令来记录每个用户的请求次数，并通过EXPIRE命令设置窗口的时间。固定窗口限流器在每个窗口结束时重置计数器，而滑动窗口限流器则在窗口开始时重置计数器。