各位观众老爷，大家好！
今天咱们来聊聊一个在互联网世界里非常重要的东西——限流。想象一下，你开了一家网红奶茶店，突然有一天，全城的人都跑来排队买你的奶茶，店里瞬间挤爆，服务器直接宕机！这就是没有限流的后果。
限流，简单来说，就是控制住涌入系统的流量，避免系统被瞬间的高流量冲垮。就像给水管装个阀门，控制水流的速度，保证水龙头不会突然爆掉。
在众多限流算法中，漏桶算法和令牌桶算法是两个非常经典且常用的。今天咱们就来深入剖析一下这哥俩，并用Redis来实现它们。
一、漏桶算法：稳如老狗，细水长流
漏桶算法，顾名思义，就像一个漏水的桶。请求就像水，不断地往桶里灌，桶以恒定的速度往外漏水（处理请求）。如果水流速度超过了漏水速度，桶就会溢出（请求被丢弃）。
漏桶算法的特点：
- 恒定速率流出： 保证了请求以固定的速度被处理，平滑了突发流量。
- 可能丢弃请求： 当请求速度超过漏水速度，桶会溢出，导致请求被丢弃，也就是限流。
- 流量整形： 将不稳定的流量整形为稳定的流量。
漏桶算法的应用场景：
- 消息队列： 保证消息以恒定的速度被消费，避免下游系统被压垮。
- API接口限流： 防止恶意攻击或者突发流量导致API接口不可用。
Redis实现漏桶算法：
我们可以用Redis的INCR和EXPIRE命令来实现漏桶算法。
INCR：原子性地增加计数器的值。EXPIRE：设置计数器的过期时间。
下面是一个简单的Python代码示例：
import redis
import time
class LeakyBucket:
def __init__(self, redis_host, redis_port, bucket_key, capacity, leak_rate):
self.redis_client = redis.Redis(host=redis_host, port=redis_port)
self.bucket_key = bucket_key
self.capacity = capacity # 桶的容量
self.leak_rate = leak_rate # 漏水速率 (每秒漏多少个请求)
def allow_request(self):
"""
判断是否允许请求通过。
"""
now = time.time()
# 首先，计算上一次漏水到现在，桶里应该漏掉了多少水
last_leak_time = float(self.redis_client.get(self.bucket_key + ":last_leak_time") or 0)
leak_amount = (now - last_leak_time) * self.leak_rate
if leak_amount > 0:
# 漏掉的水不能超过当前桶里的水量
remaining_water = float(self.redis_client.get(self.bucket_key) or 0)
new_water = max(0, remaining_water - leak_amount)
self.redis_client.set(self.bucket_key, new_water)
self.redis_client.set(self.bucket_key + ":last_leak_time", now)
# 尝试往桶里加水
current_water = float(self.redis_client.get(self.bucket_key) or 0)
if current_water + 1 <= self.capacity:
# 加水成功
self.redis_client.incr(self.bucket_key)
self.redis_client.set(self.bucket_key + ":last_leak_time", time.time()) #更新最后漏水时间
return True
else:
# 桶满了，拒绝请求
return False
# 示例用法
if __name__ == '__main__':
bucket = LeakyBucket(redis_host='localhost', redis_port=6379, bucket_key='my_bucket', capacity=10, leak_rate=1)
for i in range(20):
if bucket.allow_request():
print(f"请求 {i+1} 允许通过")
else:
print(f"请求 {i+1} 被拒绝")
time.sleep(0.2) # 模拟请求间隔代码解释：
LeakyBucket类： 定义了一个漏桶类，包含了Redis客户端、桶的key、桶的容量和漏水速率等属性。allow_request()方法： 这是核心方法，判断是否允许请求通过。- 首先，它计算从上次漏水到现在，桶里应该漏掉了多少水，并更新桶里的水量。
- 然后，它尝试往桶里加水，如果桶未满，则加水成功，允许请求通过；否则，拒绝请求。
redis_client.get(self.bucket_key + ":last_leak_time") or 0: 获取上次漏水时间，如果不存在，则初始化为0。- 示例用法： 创建一个漏桶实例，并模拟20个请求，判断每个请求是否允许通过。
漏桶算法的优缺点：
- 优点： 实现简单，能够平滑流量，保证系统稳定。
- 缺点： 无法应对突发流量，即使系统有能力处理，也会被限流。
二、令牌桶算法：积攒实力，伺机而动
令牌桶算法，顾名思义，就像一个装满令牌的桶。系统以恒定的速度往桶里放入令牌，每个请求需要从桶里获取一个令牌才能通过。如果桶里没有令牌，请求就会被拒绝。
令牌桶算法的特点：
- 允许突发流量： 只要桶里有足够的令牌，就可以处理突发流量。
- 平均速率限制： 令牌以恒定的速度生成，保证了平均速率的限制。
- 更灵活： 可以根据业务需求调整令牌生成速率和桶的容量。
令牌桶算法的应用场景：
- API接口限流： 允许一定程度的突发流量，提高用户体验。
- 消息队列： 允许消息队列在短时间内处理大量的消息。
Redis实现令牌桶算法：
我们可以用Redis的INCR和TTL命令来实现令牌桶算法。
INCR：原子性地增加计数器的值（令牌数量）。TTL：获取计数器的剩余生存时间（用于计算应该补充多少令牌）。
下面是一个简单的Python代码示例：
import redis
import time
class TokenBucket:
def __init__(self, redis_host, redis_port, bucket_key, capacity, fill_rate):
self.redis_client = redis.Redis(host=redis_host, port=redis_port)
self.bucket_key = bucket_key
self.capacity = capacity # 桶的容量
self.fill_rate = fill_rate # 令牌填充速率 (每秒填充多少个令牌)
self.lock_key = bucket_key + ":lock" # 分布式锁key，防止并发问题
def allow_request(self):
"""
判断是否允许请求通过。
"""
with self.redis_client.lock(self.lock_key, timeout=5, blocking_timeout=1): # 加锁，防止并发问题
now = time.time()
# 上次请求的时间
last_refill_time = float(self.redis_client.get(self.bucket_key + ":last_refill_time") or now)
# 计算这段时间应该补充的令牌数量
refill_amount = (now - last_refill_time) * self.fill_rate
if refill_amount > 0:
# 将令牌补充到桶里
current_tokens = float(self.redis_client.get(self.bucket_key) or 0)
new_tokens = min(self.capacity, current_tokens + refill_amount)
self.redis_client.set(self.bucket_key, new_tokens)
self.redis_client.set(self.bucket_key + ":last_refill_time", now)
# 尝试获取令牌
current_tokens = float(self.redis_client.get(self.bucket_key) or 0)
if current_tokens >= 1:
# 获取令牌成功
self.redis_client.decr(self.bucket_key)
return True
else:
# 没有令牌，拒绝请求
return False
# 示例用法
if __name__ == '__main__':
bucket = TokenBucket(redis_host='localhost', redis_port=6379, bucket_key='my_token_bucket', capacity=10, fill_rate=2)
for i in range(20):
if bucket.allow_request():
print(f"请求 {i+1} 允许通过")
else:
print(f"请求 {i+1} 被拒绝")
time.sleep(0.1) # 模拟请求间隔代码解释：
TokenBucket类： 定义了一个令牌桶类，包含了Redis客户端、桶的key、桶的容量和令牌填充速率等属性。allow_request()方法： 这是核心方法，判断是否允许请求通过。- 首先，它计算从上次填充令牌到现在，应该补充多少令牌，并更新桶里的令牌数量。
- 然后，它尝试从桶里获取一个令牌，如果桶里有令牌，则获取令牌成功，允许请求通过；否则，拒绝请求。
redis_client.get(self.bucket_key + ":last_refill_time") or now: 获取上次填充令牌时间，如果不存在，则初始化为当前时间。redis_client.lock(self.lock_key, timeout=5, blocking_timeout=1): 使用Redis的分布式锁，防止并发问题，保证令牌桶的原子性操作。- 示例用法： 创建一个令牌桶实例，并模拟20个请求，判断每个请求是否允许通过。
令牌桶算法的优缺点：
- 优点： 允许突发流量，更灵活，可以根据业务需求调整参数。
- 缺点： 实现相对复杂，需要考虑并发问题。
三、漏桶 vs 令牌桶：一场友谊赛
| 特性 | 漏桶算法 | 令牌桶算法 |
|---|---|---|
| 处理流量 | 恒定速率 | 允许突发流量 |
| 实现难度 | 简单 | 相对复杂 |
| 应用场景 | 对流量平滑要求高的场景 | 需要处理突发流量的场景 |
| 丢弃请求 | 当桶溢出时丢弃请求 | 当没有令牌时丢弃请求 |
| 流量整形 | 将不稳定的流量整形为稳定的流量 | 限制平均速率，允许一定程度的突发 |
总结：
- 漏桶算法： 就像一个严格的门卫，只允许固定数量的人进入，保证内部秩序井然。
- 令牌桶算法： 就像一个宽松的门卫，只要你有通行证（令牌），就可以进入，允许偶尔的拥挤。
选择哪种算法，取决于你的具体业务需求。如果你的系统需要绝对的稳定，不允许任何突发流量，那么漏桶算法更适合你。如果你的系统需要一定的灵活性，允许一定程度的突发流量，那么令牌桶算法更适合你。
四、更高级的限流策略：组合拳出击
在实际应用中，我们往往需要更复杂的限流策略，例如：
- 基于用户ID的限流： 限制每个用户的请求频率，防止恶意用户滥用。
- 基于IP地址的限流： 限制来自特定IP地址的请求频率，防止恶意攻击。
- 基于API接口的限流： 针对不同的API接口设置不同的限流策略。
这些更高级的限流策略，可以在漏桶算法和令牌桶算法的基础上进行扩展，例如：
- 多级漏桶： 使用多个漏桶，每个漏桶处理不同类型的请求。
- 滑动窗口： 使用滑动窗口来统计请求数量，更加精确地控制请求速率。
五、最后的温馨提示：
- 选择合适的Redis数据结构： 除了
INCR和EXPIRE，还可以使用Redis的Sorted Set来实现更复杂的限流逻辑。 - 注意并发问题： 在高并发场景下，一定要注意并发问题，可以使用Redis的分布式锁来保证原子性操作。
- 监控和告警： 监控限流器的运行状态，及时发现和解决问题。
- 根据业务需求调整参数： 不同的业务场景需要不同的限流参数，要根据实际情况进行调整。
好了，今天的分享就到这里。希望大家能够掌握漏桶算法和令牌桶算法的精髓，并在实际工作中灵活运用，保护好自己的系统，避免被流量冲垮！
记住，限流不是目的，而是手段，目的是为了保证系统的稳定性和可用性，最终为用户提供更好的服务。
感谢大家的观看！我们下期再见！