redis限流通过控制单位时间请求量保护系统，常见算法有计数器、令牌桶和漏桶。1.计数器算法简单但存在临界问题，适合对精度要求不高的场景；2.令牌桶允许突发流量，通过恒定填充令牌实现更平滑的限流；3.漏桶以固定速率处理请求，严格限制流量但无法应对突发流量。选择时需根据业务需求权衡实现复杂度与限流效果，并结合redis集群或分布式锁解决性能与一致性问题，从而构建稳定可靠的系统。
Redis限流，本质上就是控制请求访问资源的速度，防止系统被突发流量冲垮。它通过限制单位时间内请求的数量来实现。
要理解Redis限流，得先明白它背后的原理。简单来说，就是利用Redis的原子性操作，比如
INCR，来记录单位时间内的请求次数。当请求次数超过预设的阈值时，就拒绝后续的请求。
Redis限流：令牌桶、漏桶、计数器，哪种更适合你？
在Redis中实现限流，常见的算法有令牌桶、漏桶和计数器。选择哪种算法，取决于你的具体需求和场景。
1. 计数器算法：简单粗暴，但有缺陷
计数器算法是最简单的限流实现方式。它通过Redis的
INCR命令，对某个键（比如用户ID）进行计数，并设置一个过期时间。每次请求到来时，计数加1，如果超过预设的阈值，就拒绝请求。
import redis
import time
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def is_allowed(user_id, limit, period):
"""
使用计数器算法进行限流
Args:
user_id: 用户ID
limit: 单位时间内允许的请求数量
period: 时间窗口，单位秒
Returns:
True: 允许请求
False: 拒绝请求
"""
key = f"limit:{user_id}"
current_count = redis_client.incr(key)
if current_count == 1:
redis_client.expire(key, period)
if current_count > limit:
return False
return True
# 示例
user_id = "user123"
limit = 5
period = 60 # 60秒内最多5次请求
for i in range(10):
if is_allowed(user_id, limit, period):
print(f"请求 {i+1} 允许")
else:
print(f"请求 {i+1} 拒绝")
time.sleep(5)优点： 实现简单，易于理解。
缺点： 存在临界问题。如果在时间窗口的末尾和下一个时间窗口的开始，都发送了接近阈值的请求，那么实际的请求数量可能会超过阈值。例如，在第59秒发送了5个请求，在第61秒又发送了5个请求，虽然每个时间窗口内都没超过5个请求，但在2秒内却发送了10个请求。
2. 令牌桶算法：更平滑的限流
令牌桶算法以恒定的速率向桶中放入令牌。每个请求到来时，需要从桶中获取一个令牌，如果获取不到，就拒绝请求。
import redis
import time
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def is_allowed_token_bucket(user_id, limit, period, fill_rate):
"""
使用令牌桶算法进行限流
Args:
user_id: 用户ID
limit: 桶的容量
period: 填充令牌的时间间隔，单位秒
fill_rate: 令牌填充速率，单位 个/秒
Returns:
True: 允许请求
False: 拒绝请求
"""
key = f"token:{user_id}"
now = time.time()
last_refill_time = redis_client.get(f"last_refill:{user_id}")
if last_refill_time is None:
last_refill_time = now
else:
last_refill_time = float(last_refill_time)
# 计算应该填充的令牌数量
refill_tokens = (now - last_refill_time) * fill_rate
if refill_tokens > 0:
# 更新桶中的令牌数量
current_tokens = redis_client.get(key)
if current_tokens is None:
current_tokens = 0
else:
current_tokens = int(current_tokens)
current_tokens = min(limit, current_tokens + refill_tokens)
redis_client.set(key, current_tokens)
redis_client.set(f"last_refill:{user_id}", now)
# 尝试获取令牌
current_tokens = redis_client.get(key)
if current_tokens is None or int(current_tokens) <= 0:
return False
else:
redis_client.decr(key)
return True
# 示例
user_id = "user456"
limit = 10
period = 1
fill_rate = 2 # 每秒填充2个令牌
for i in range(20):
if is_allowed_token_bucket(user_id, limit, period, fill_rate):
print(f"请求 {i+1} 允许")
else:
print(f"请求 {i+1} 拒绝")
time.sleep(0.2)优点： 允许一定程度的突发流量，因为桶中可以存储一定数量的令牌。
缺点： 实现相对复杂，需要维护令牌桶的状态。
3. 漏桶算法：更严格的限流
漏桶算法以恒定的速率从桶中漏出请求。请求先进入桶中，如果桶满了，就拒绝请求。
import redis
import time
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def is_allowed_leaky_bucket(user_id, limit, rate):
"""
使用漏桶算法进行限流
Args:
user_id: 用户ID
limit: 桶的容量
rate: 漏水速率，单位 个/秒
Returns:
True: 允许请求
False: 拒绝请求
"""
key = f"bucket:{user_id}"
bucket_size = redis_client.llen(key)
if bucket_size < limit:
redis_client.lpush(key, time.time()) # 将当前时间戳放入桶中
redis_client.expire(key, limit / rate + 1) # 设置过期时间，防止桶无限增长
# 移除过期的请求
while True:
oldest_request_time = redis_client.rpop(key)
if oldest_request_time is None:
break
if time.time() - float(oldest_request_time) < limit / rate:
redis_client.rpush(key, oldest_request_time) # 重新放回桶中
break
return True
else:
return False
# 示例
user_id = "user789"
limit = 5 # 桶的容量
rate = 1 # 每秒漏出1个请求
for i in range(10):
if is_allowed_leaky_bucket(user_id, limit, rate):
print(f"请求 {i+1} 允许")
else:
print(f"请求 {i+1} 拒绝")
time.sleep(0.5)优点： 可以平滑流量，保证请求以恒定的速率被处理。
缺点： 无法处理突发流量，因为桶的容量是有限的。
如何选择合适的Redis限流算法？
选择哪种限流算法，需要根据具体的业务场景来决定。
- 如果需要简单快速的实现，且对流量平滑性要求不高，可以选择计数器算法。
- 如果需要允许一定程度的突发流量，可以选择令牌桶算法。
- 如果需要严格控制请求的速率，保证请求以恒定的速率被处理，可以选择漏桶算法。
Redis限流的常见问题与解决方案
在使用Redis限流的过程中，可能会遇到一些问题，比如：
- Redis性能瓶颈： 如果限流的请求量非常大，可能会导致Redis的性能瓶颈。可以考虑使用Redis集群，或者使用更高效的限流算法。
- 数据一致性问题： 在分布式环境下，需要保证限流数据的一致性。可以使用Redis的分布式锁，或者使用CAP理论中的CP系统。
- 误判问题： 由于网络延迟等原因，可能会导致限流算法误判。可以适当调整限流的阈值，或者使用更复杂的限流算法。
总的来说，Redis限流是一种非常有效的保护系统的方法。选择合适的限流算法，并解决可能遇到的问题，可以帮助你构建更稳定、更可靠的系统。
