Leaky Bucket 漏桶算法：强制恒定速率

Leaky Bucket 漏桶算法入门：一文读懂流量整形核心机制

在构建高可用系统与网络流量控制时，“限流”是绕不开的话题。Leaky Bucket（漏桶） 是最经典的流量整形（Traffic Shaping）算法之一，其核心思想可以概括为：强制恒定输出速率，抵抗突发流量。本文将从零开始，带你理解漏桶的原理、工作流程、实现方式以及适用场景，即使没有深厚的计算机基础也能读懂。

1. 什么是漏桶算法？

想象一个底部开了一个小孔的水桶。无论你以多快的速度往桶里倒水（输入流量），水只会以固定的速率从底部小孔流出（输出流量）。如果倒水的速度长期大于流出的速度，桶就会被装满，之后多余的水会直接溢出（丢弃数据包或拒绝请求）。

把这个比喻映射到计算机系统：

• 水桶：代表一个固定容量的缓冲区（队列）。
• 倒水：代表用户请求到达系统。
• 恒定流出：代表系统以固定的速率处理请求。
• 水满溢出：代表当缓冲区满时，新到达的请求会被直接丢弃。

一句话总结：漏桶算法强制将不均匀的输入流量转化为平滑、恒定速率的输出流量，保护后端服务不被突发流量冲垮。

2. 核心目标与应用场景

漏桶算法的设计目标不是“尽可能多地处理请求”，而是严格平滑流量突发，使流量变得可预测。它主要用于以下场景：

• 网络流量管理：在网络路由器或流量整形器中，将突发数据包转换为恒定的比特流，避免下游设备过载。
• API 网关限流：保护后端微服务，确保每秒请求数（RPS）稳定，防止数据库或计算资源耗尽。
• 消息队列消费：消费者按固定速率拉取消息，避免处理速度波动导致背压。
• 防止恶意刷接口：强制每个用户以稳定低速访问，超出直接拒绝，减轻服务器压力。

如果在你的系统中，稳定性与可预测性远比“突然的高吞吐”重要，漏桶就是非常合适的选择。

3. 漏桶的数学模型与工作流程

一个标准的漏桶算法需要以下三个参数：

处理每一次新请求的流程：

1. 计算漏水：根据上次漏水的时间戳与当前时间，计算出在此时间段内应漏出的请求数 leak = (currentTime - lastTime) × R。
2. 更新水位：W = max(0, W - leak)。水位不能为负。
3. 判断溢出：如果 W + 1 <= B（假设每个请求占用1个单位），则请求入桶：W = W + 1，返回“允许通过”；否则返回“请求被丢弃”。
4. 记录时间：更新 lastTime 为当前时间。

注意：漏桶的“平滑”特性体现在：即使某一刻有 100 个请求同时到达，只要 W + 100 <= B，它们都会被加入桶中；但桶会以固定速率 R 逐个释放，呈现出匀速的处理效果。这就是流量整形——削峰填谷。

4. 与令牌桶（Token Bucket）的对比

漏桶经常被拿来与另一个著名算法令牌桶（Token Bucket） 比较。很多初学者容易混淆，这里用一张表清晰区分：

简单记忆：漏桶“强制排队，匀速处理”；令牌桶“预支额度，允许突击消费”。如果你想让接口在任何时刻的处理速率都绝不超限，选漏桶；如果希望允许短时间尖峰，再用令牌桶。

5. 动手实现一个简单的漏桶（Python 伪代码）

为了让概念落地，下面用 Python 实现一个线程安全的漏桶类。你只需理解面向对象的思维方式，就能直接改进用于生产环境。

import time
import threading

class LeakyBucket:
    def __init__(self, capacity: int, rate: float):
        """
        capacity: 桶的最大容量（可容纳的最大请求数）
        rate:     漏出速率（请求/秒）
        """
        self.capacity = capacity
        self.rate = rate
        self.water = 0               # 当前水位
        self.last_leak_time = time.monotonic()
        self.lock = threading.Lock()

    def _leak(self):
        """根据时间推移，减少水位（漏水）"""
        now = time.monotonic()
        elapsed = now - self.last_leak_time
        leaked = int(elapsed * self.rate)
        if leaked > 0:
            self.water = max(0, self.water - leaked)
            self.last_leak_time = now

    def allow_request(self) -> bool:
        """尝试放进一个请求，返回True表示允许，False表示丢弃"""
        with self.lock:
            self._leak()
            if self.water < self.capacity:
                self.water += 1
                return True
            else:
                return False

使用示例：

# 桶容量为5，处理速率每秒2个请求
bucket = LeakyBucket(capacity=5, rate=2.0)

for i in range(10):
    if bucket.allow_request():
        print(f"请求{i} 被处理")
    else:
        print(f"请求{i} 被丢弃（桶已满）")
    time.sleep(0.1)   # 模拟请求间隔

运行解释：桶在初始化后，以每秒 2 个的速度漏水。由于请求间隔很短，前几个请求会填满桶，之后溢出的请求直接被丢弃。被接受的请求实际上会被“排队”处理，保证了处理速率的恒定。

6. 漏桶算法的优缺点

优点：

• 严格的输出速率：为下游系统提供了极其稳定的负载，避免瞬间压力。
• 实现直观：基于队列和定时器，概念简单。
• 自然抵抗流量脉冲：超额的请求直接被拒绝，不会积压到后端。

缺点：

• 缺乏弹性：无法利用系统空闲时的资源去“预积累”处理能力，一旦进入高负载，所有多余请求都被丢弃，即使下游短暂有能力处理更多。
• 可能浪费资源：当请求速率长期低于处理速率时，桶一直是空的，看起来没有用武之地；但其存在的价值在于防护不可预知的突发。
• 参数调优敏感：桶容量太小会导致正常微突发被误杀；太大则失去平滑效果，增加内存消耗。

7. 生产环境中使用漏桶的注意事项

• 选择合适的速率单位：对于网络数据包，速率可能是“比特/秒”或“字节/秒”；对于 API 请求，是“请求/秒”。确保单位与业务一致。
• 多级漏桶联动：在复杂网关中，可以针对不同的用户、接口或 IP 实现独立的漏桶，形成多级限流矩阵。
• 监控与告警：记录满桶丢弃的次数和当前水位，帮助你判断参数是否合理，并及时调整容量 B 或速率 R。
• 分布式环境：如果服务部署在多个节点，需要中心化存储（如 Redis）来维护统一的桶状态，否则全局限流会失效。

8. 扩展阅读

本文只聚焦了漏桶的经典形态。如果你已经掌握了它的精髓，可以进一步探索：

• 自适应漏桶：根据下游实时负载动态调整速率 R，在保证稳定的同时提高资源利用率。
• 加权漏桶：对不同优先级的请求使用不同的“水位增加系数”，实现有区分的丢弃策略。
• 结合模式：在实际项目中，漏桶常与计数器固定窗口、滑动窗口日志等组合成混合限流器，以应对多维度的限制需求。

理解漏桶不仅是学会一个算法，更是掌握流量整形思想的开始。希望这篇教程为你打下扎实的基础。欢迎收藏我们的“免费在线教程”，持续更新更多网络与系统设计经典内容。