redis实现限流的常见策略

Redis实现限流的常见策略包括：

1. 固定窗口限流器（Fixed Window Rate Limiter）
2. 滑动窗口限流器（Sliding Window Rate Limiter）
3. 漏桶算法（Leaky Bucket Algorithm）
4. 令牌桶算法（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命令设置窗口的时间。固定窗口限流器在每个窗口结束时重置计数器，而滑动窗口限流器则在窗口开始时重置计数器。