深入 Rate Limiting（限流）：漏桶算法 vs 令牌桶算法，在高并发突发流量下的表现差异

各位技术同仁，大家好！

今天，我们将深入探讨一个在构建高可用、高性能分布式系统时至关重要的技术：Rate Limiting，即限流。在微服务架构盛行，API经济蓬勃发展的今天，如何保护我们的服务不受突发流量冲击，保障系统稳定运行，同时提供公平的资源访问，限流机制扮演着举足轻重的作用。我们将聚焦两种最经典、最广泛使用的限流算法：漏桶算法（Leaky Bucket）与令牌桶算法（Token Bucket），并详细分析它们在高并发突发流量下的表现差异。

1. 限流的必要性与核心目标

想象一下，你精心设计的API服务，平时运行良好，但在某个热门事件、促销活动或恶意攻击下，瞬间涌入数倍甚至数十倍的请求。如果没有限流机制，会发生什么？

• 系统过载崩溃： 服务器CPU、内存、网络IO瞬间飙升，服务响应变慢甚至宕机，导致雪崩效应。
• 资源滥用： 少数用户或服务可能耗尽所有资源，导致其他正常用户无法访问。
• 成本失控： 云服务按量计费，突发流量可能导致意外的高昂费用。
• 服务质量下降： 用户体验变差，请求延迟增加，甚至大量请求失败。

限流的核心目标，正是为了解决这些问题，它像一道智能的闸门，控制着流入我们系统的请求速率，确保系统能够持续、稳定、可预测地对外提供服务。

2. 限流算法的基石

限流算法的核心思想是，在单位时间内只允许一定数量的请求通过。当请求速率超过预设阈值时，多余的请求会被拒绝、排队或降级处理。常见的限流算法包括：

• 固定窗口（Fixed Window）
• 滑动窗口（Sliding Window）
• 漏桶算法（Leaky Bucket）
• 令牌桶算法（Token Bucket）

今天，我们的焦点是后两者，它们在实际应用中尤其常见，各有千秋。

3. 漏桶算法：平滑流量的守护者

漏桶算法是一个经典的流量整形算法，它的设计哲学是：无论入口流量有多大，出口流量都将保持恒定速率。

3.1 核心思想与类比

想象一个底部有孔的桶，水以恒定的速率从孔中流出。我们可以向桶中加水（代表请求），但如果加水速度过快，桶满溢出，那么多余的水（请求）就会被直接丢弃。

• 桶（Bucket）： 存储请求的队列，有固定容量。
• 水（Water）： 进入桶的请求。
• 漏水速率（Leak Rate）： 桶底部的孔以固定速率排出水，代表系统处理请求的速率。

3.2 运行机制

1. 请求进入： 当一个请求到来时，它会尝试加入漏桶。
2. 容量检查： 如果漏桶未满，请求被放入桶中（即加入队列）。
3. 漏水处理： 漏桶以预设的固定速率将请求从桶中取出并处理。
4. 溢出拒绝： 如果漏桶已满，新来的请求将被直接拒绝或丢弃。

这种机制强制所有通过漏桶的请求都以一个恒定的速率进行处理，从而达到平滑流量的效果。

3.3 漏桶算法的特点

• 强制平滑输出： 这是漏桶算法最显著的特点。即使输入流量是突发性的，输出流量也会被整形为恒定速率。
• 排队机制： 漏桶内部通常维护一个请求队列，允许在一定程度上缓冲突发请求。
• 丢弃溢出： 当队列满时，新来的请求会被直接拒绝，防止系统过载。
• 适用于： 需要严格控制后端服务处理速率的场景，例如数据库写入、消息队列发送、敏感资源访问等。

3.4 Python 实现示例

下面是一个简化的漏桶算法的 Python 实现，我们使用 threading.Lock 来保证在多线程环境下的线程安全。

import time
import collections
import threading

class LeakyBucket:
    """
    漏桶算法限流器
    """
    def __init__(self, capacity: int, leak_rate: float):
        """
        初始化漏桶。
        :param capacity: 桶的容量，即最多能排队多少个请求。
        :param leak_rate: 漏水速率，单位是请求/秒。
                          例如，leak_rate=2 表示每秒处理2个请求。
        """
        if capacity <= 0 or leak_rate <= 0:
            raise ValueError("Capacity and leak rate must be positive.")

        self.capacity = capacity  # 桶的容量
        self.leak_rate = leak_rate # 漏水速率 (请求/秒)
        self.current_water = 0.0 # 当前桶中的水量，可以是浮点数以更精确计算
        self.last_leak_time = time.time() # 上次计算漏水的时间戳
        self.lock = threading.Lock() # 线程安全锁

    def _update_water(self):
        """
        根据时间流逝更新桶中的水量（模拟漏水）。
        这个方法在每次尝试获取请求前调用，确保水量是最新的。
        """
        now = time.time()
        time_elapsed = now - self.last_leak_time

        # 计算这段时间流出的水量
        leaked_amount = time_elapsed * self.leak_rate

        # 更新当前水量，确保不小于0
        self.current_water = max(0.0, self.current_water - leaked_amount)

        # 更新上次漏水时间
        self.last_leak_time = now

    def try_acquire(self) -> bool:
        """
        尝试获取一个请求令牌。
        :return: 如果成功获取（请求被接受），返回True；否则（桶满或拒绝），返回False。
        """
        with self.lock:
            # 首先更新桶中水量，模拟漏水
            self._update_water()

            # 判断当前水量是否小于容量，如果小于，则可以放入新请求
            if self.current_water < self.capacity:
                self.current_water += 1.0 # 放入一个请求
                return True
            else:
                # 桶已满，拒绝请求
                return False

    def get_status(self):
        """
        获取当前漏桶状态（仅用于调试）。
        """
        with self.lock:
            self._update_water() # 获取最新状态前先更新
            return {
                "capacity": self.capacity,
                "leak_rate": self.leak_rate,
                "current_water": self.current_water,
                "last_leak_time": self.last_leak_time
            }

# --- 漏桶算法在高并发突发流量下的表现模拟 ---
def simulate_leaky_bucket(bucket: LeakyBucket, request_count: int, burst_size: int, burst_interval: float):
    print(f"n--- 模拟漏桶算法 (容量: {bucket.capacity}, 漏速: {bucket.leak_rate} req/s) ---")
    print(f"将发送 {request_count} 个请求，其中包含每 {burst_interval} 秒一次的 {burst_size} 个请求的突发。")

    granted_count = 0
    denied_count = 0
    start_time = time.time()

    # 模拟一个突发队列，快速发送请求
    burst_requests = []
    for i in range(burst_size):
        burst_requests.append(i)

    for i in range(request_count):
        if i > 0 and i % burst_size == 0: # 模拟每隔一段时间的突发
            print(f"n--- 突发流量开始 (第 {i//burst_size + 1} 次突发) ---")
            for j in range(burst_size):
                if bucket.try_acquire():
                    granted_count += 1
                    # print(f"[{time.time() - start_time:.3f}s] Request {i+j+1}: GRANTED. Current water: {bucket.get_status()['current_water']:.2f}")
                else:
                    denied_count += 1
                    # print(f"[{time.time() - start_time:.3f}s] Request {i+j+1}: DENIED (Bucket full). Current water: {bucket.get_status()['current_water']:.2f}")
            print(f"--- 突发流量结束 ---")
            time.sleep(burst_interval) # 突发后等待一段时间
        else:
            # 正常流量
            if bucket.try_acquire():
                granted_count += 1
                # print(f"[{time.time() - start_time:.3f}s] Request {i+1}: GRANTED. Current water: {bucket.get_status()['current_water']:.2f}")
            else:
                denied_count += 1
                # print(f"[{time.time() - start_time:.3f}s] Request {i+1}: DENIED (Bucket full). Current water: {bucket.get_status()['current_water']:.2f}")
            time.sleep(0.05) # 模拟请求间隔

    end_time = time.time()
    print(f"n模拟结束。总耗时: {end_time - start_time:.3f} 秒")
    print(f"总请求数: {request_count + (request_count // burst_size) * burst_size}") # 考虑突发请求的实际数量
    print(f"被允许请求数: {granted_count}")
    print(f"被拒绝请求数: {denied_count}")

# 示例：漏桶限流器
# 容量为 5，漏水速率为每秒 2 个请求
lb = LeakyBucket(capacity=5, leak_rate=2) 
# 模拟发送 50 个请求，每 2 秒模拟一次 10 个请求的突发
# simulate_leaky_bucket(lb, request_count=50, burst_size=10, burst_interval=2)

漏桶在高并发突发流量下的表现：

当突发流量来临时，如果桶中有足够的空间，请求会被放入桶中排队。然而，一旦桶被填满，后续的突发请求将立即被拒绝。漏桶算法会不遗余力地以恒定速率处理请求，哪怕这意味着在短时间内拒绝大量涌入的请求。它的目标是保护后端服务，而不是尽可能多地处理突发请求。因此，在突发流量下，漏桶可能会导致较高的请求拒绝率，但能够保证后端服务的负载稳定。

4. 令牌桶算法：允许突发但限制平均速率

令牌桶算法与漏桶算法恰恰相反，它允许在短时间内处理比平均速率更高的请求，即允许突发，但长期平均速率仍受限制。

4.1 核心思想与类比

想象一个固定容量的桶，系统会以恒定速率往桶中放置令牌。每个请求在被处理之前，必须从桶中取走一个令牌。如果桶中没有令牌，请求就必须等待，或者被拒绝。

• 桶（Bucket）： 存储令牌的容器，有固定容量。
• 令牌（Token）： 系统以恒定速率生成并放入桶中。每个请求消耗一个令牌。
• 填充速率（Fill Rate）： 令牌生成并放入桶中的速率。

4.2 运行机制

1. 令牌生成： 系统以预设的固定速率向令牌桶中添加令牌，直到桶满。
2. 请求到来： 当一个请求到来时，它会尝试从令牌桶中获取一个令牌。
3. 获取令牌：
   • 如果桶中有足够的令牌，请求会取走一个令牌并被处理。
   • 如果桶中没有令牌，请求将被拒绝或等待，直到有新的令牌生成。
4. 桶容量： 桶的容量决定了允许的最大突发请求数量。

4.3 令牌桶算法的特点

• 允许突发： 令牌桶可以积累令牌，因此在短时间内，如果桶中有足够的令牌，可以处理超过平均速率的请求，这正是其与漏桶最大的区别。
• 限制平均速率： 令牌的生成速率决定了长期平均可处理的请求速率。
• 弹性： 相较于漏桶，令牌桶在应对流量波动时更具弹性。
• 适用于： 用户API、需要有一定突发能力的场景，例如允许用户在短时间内发送更多请求，但总的请求量仍然受控。

4.4 Python 实现示例

下面是一个简化的令牌桶算法的 Python 实现。

import time
import threading

class TokenBucket:
    """
    令牌桶算法限流器
    """
    def __init__(self, capacity: int, fill_rate: float):
        """
        初始化令牌桶。
        :param capacity: 桶的容量，即最多能存储多少个令牌。这决定了最大突发量。
        :param fill_rate: 令牌填充速率，单位是令牌/秒。这决定了长期平均处理速率。
                          例如，fill_rate=5 表示每秒生成5个令牌。
        """
        if capacity <= 0 or fill_rate <= 0:
            raise ValueError("Capacity and fill rate must be positive.")

        self.capacity = capacity  # 桶的容量
        self.fill_rate = fill_rate # 令牌填充速率 (令牌/秒)
        self.current_tokens = float(capacity) # 当前桶中的令牌数量，初始为满
        self.last_fill_time = time.time() # 上次填充令牌的时间戳
        self.lock = threading.Lock() # 线程安全锁

    def _add_tokens(self):
        """
        根据时间流逝向桶中添加令牌。
        这个方法在每次尝试获取请求前调用，确保令牌数量是最新的。
        """
        now = time.time()
        time_elapsed = now - self.last_fill_time

        # 计算这段时间生成的令牌数量
        tokens_to_add = time_elapsed * self.fill_rate

        # 更新当前令牌数量，确保不超过桶容量
        self.current_tokens = min(self.capacity, self.current_tokens + tokens_to_add)

        # 更新上次填充时间
        self.last_fill_time = now

    def try_acquire(self, num_tokens: int = 1) -> bool:
        """
        尝试获取指定数量的令牌。
        :param num_tokens: 尝试获取的令牌数量，默认为1。
        :return: 如果成功获取令牌（请求被接受），返回True；否则（令牌不足），返回False。
        """
        with self.lock:
            # 首先更新桶中令牌数量
            self._add_tokens()

            # 判断当前令牌数量是否足够
            if self.current_tokens >= num_tokens:
                self.current_tokens -= num_tokens # 消耗令牌
                return True
            else:
                # 令牌不足，拒绝请求
                return False

    def get_status(self):
        """
        获取当前令牌桶状态（仅用于调试）。
        """
        with self.lock:
            self._add_tokens() # 获取最新状态前先更新
            return {
                "capacity": self.capacity,
                "fill_rate": self.fill_rate,
                "current_tokens": self.current_tokens,
                "last_fill_time": self.last_fill_time
            }

# --- 令牌桶算法在高并发突发流量下的表现模拟 ---
def simulate_token_bucket(bucket: TokenBucket, request_count: int, burst_size: int, burst_interval: float):
    print(f"n--- 模拟令牌桶算法 (容量: {bucket.capacity}, 填充速率: {bucket.fill_rate} req/s) ---")
    print(f"将发送 {request_count} 个请求，其中包含每 {burst_interval} 秒一次的 {burst_size} 个请求的突发。")

    granted_count = 0
    denied_count = 0
    start_time = time.time()

    for i in range(request_count):
        if i > 0 and i % burst_size == 0: # 模拟每隔一段时间的突发
            print(f"n--- 突发流量开始 (第 {i//burst_size + 1} 次突发) ---")
            for j in range(burst_size):
                if bucket.try_acquire():
                    granted_count += 1
                    # print(f"[{time.time() - start_time:.3f}s] Request {i+j+1}: GRANTED. Current tokens: {bucket.get_status()['current_tokens']:.2f}")
                else:
                    denied_count += 1
                    # print(f"[{time.time() - start_time:.3f}s] Request {i+j+1}: DENIED (No tokens). Current tokens: {bucket.get_status()['current_tokens']:.2f}")
            print(f"--- 突发流量结束 ---")
            time.sleep(burst_interval) # 突发后等待一段时间
        else:
            # 正常流量
            if bucket.try_acquire():
                granted_count += 1
                # print(f"[{time.time() - start_time:.3f}s] Request {i+1}: GRANTED. Current tokens: {bucket.get_status()['current_tokens']:.2f}")
            else:
                denied_count += 1
                # print(f"[{time.time() - start_time:.3f}s] Request {i+1}: DENIED (No tokens). Current tokens: {bucket.get_status()['current_tokens']:.2f}")
            time.sleep(0.05) # 模拟请求间隔

    end_time = time.time()
    print(f"n模拟结束。总耗时: {end_time - start_time:.3f} 秒")
    print(f"总请求数: {request_count + (request_count // burst_size) * burst_size}")
    print(f"被允许请求数: {granted_count}")
    print(f"被拒绝请求数: {denied_count}")

# 示例：令牌桶限流器
# 容量为 10，填充速率为每秒 5 个令牌
tb = TokenBucket(capacity=10, fill_rate=5) 
# 模拟发送 50 个请求，每 2 秒模拟一次 10 个请求的突发
# simulate_token_bucket(tb, request_count=50, burst_size=10, burst_interval=2)

令牌桶在高并发突发流量下的表现：

当突发流量来临时，如果令牌桶中有足够的令牌（因为在流量低谷时积累了一些），它能够以高于平均速率的速度处理这批请求，直到令牌耗尽。这使得令牌桶在应对瞬时高并发时表现得更为友好，能够“吸收”一部分流量冲击。只有当突发流量持续时间过长，或者突发量远超桶的容量时，请求才会被拒绝。因此，令牌桶在突发流量下的拒绝率通常会低于漏桶，但代价是后端服务在短时间内可能会承受更高的瞬时负载。

5. 漏桶 vs 令牌桶：高并发突发流量下的表现差异

现在，让我们通过一个表格来直观地对比这两种算法的核心特性和在高并发突发流量下的具体表现。

特性/算法	漏桶算法（Leaky Bucket）	令牌桶算法（Token Bucket）
核心思想	强制输出速率恒定，平滑流量。	限制平均速率，允许突发。
处理方式	请求入队，按固定速率出队。队列满则拒绝。	令牌入桶，请求消耗令牌。无令牌则拒绝。
流量整形	是：将任何突发流量整形为平滑的固定速率输出。	否：不强制平滑输出，允许输出速率波动。
突发处理能力	差：桶满后立即拒绝，不具备应对突发的能力。	好：可积累令牌，在短时间内处理高于平均速率的请求。
高并发下行为	请求会被排队或拒绝。后端服务负载稳定，但用户体验可能因延迟或拒绝而受损。	可快速处理积攒的令牌，后端服务在短时间内承受瞬时高峰，但总请求量受限。
突发流量下行为	桶满则拒绝大量请求，拒绝率高。后端服务负载维持在设定阈值。	可根据桶容量处理一定量的突发，拒绝率相对较低。突发结束后，处理速率回归平均。
参数	容量（队列大小）、漏水速率（处理速率）。	容量（令牌最大数）、填充速率（令牌生成速率）。
适用场景	需严格控制后端服务负载，对延迟不敏感，或后端处理能力有限且稳定。如：消息队列、数据库写入、系统内部服务间限流。	需允许一定程度的突发，提供更灵活的用户体验。如：用户API接口、对外开放的第三方API。
实时性	较差，可能导致请求长时间排队。	较好，能更快响应突发请求。

5.1 详细分析在高并发突发流量下的表现

漏桶算法的视角：

当大量请求在短时间内涌入（高并发突发流量）时，漏桶算法会像一个坚定的守门员。

• 队列缓冲： 如果漏桶的容量（队列）足够大，它会先将一部分请求放入队列中等待。
• 固定速率处理： 无论入口流量有多大，出口流量（即实际处理的请求）始终保持在 leak_rate。
• 快速拒绝： 一旦漏桶中的水量达到容量上限，所有新来的请求都会被立即拒绝。它不会等到系统崩溃才拒绝，而是在达到预设的“安全水位”时就果断拒绝。

结果： 在突发流量下，漏桶算法能够极其有效地保护后端服务，使其负载保持在一个恒定的、可控的水平。但代价是，用户可能会经历较高的请求延迟（如果请求在队列中等待）或较高的请求拒绝率。对于那些对实时性要求高、但后端服务处理能力有限的场景，漏桶算法的这种“宁可错杀一千，不可放过一个”的策略可能导致用户体验不佳。

令牌桶算法的视角：

当高并发突发流量来临时，令牌桶算法则显得更为灵活和“人性化”。

• 吸收突发： 如果在突发到来之前，令牌桶已经积累了足够的令牌（因为在流量低谷时，令牌会持续生成并存储），那么这些突发请求可以迅速消耗掉这些令牌，从而在短时间内以远超 fill_rate 的速度被处理。这就是所谓的“突发能力”。
• 平稳过渡： 一旦桶中的令牌被消耗殆尽，后续的请求就必须等待新的令牌生成（即回归到 fill_rate 的平均处理速率），或者被拒绝。
• 延迟拒绝： 令牌桶不会像漏桶那样一满就立即拒绝，它会先尝试消耗积累的“信用”（令牌）。

结果： 在突发流量下，令牌桶算法能够更好地应对短期的流量冲击，减少请求的拒绝率，提供更平滑的用户体验。后端服务在短时间内可能会承受一个高于平均水平的瞬时负载，但这个负载的峰值是由令牌桶的容量 (capacity) 和令牌消耗速度决定的，而不是无限制的。当突发过去，令牌生成速率将再次限制平均处理速率。

5.2 总结性对比

可以说，漏桶算法更侧重于对后端服务的保护，确保其处理能力不会被压垮，牺牲了对突发流量的友好性。它是一个“平滑器”，强制输出稳定。

而令牌桶算法更侧重于对前端请求的响应和用户体验，允许在不影响长期平均速率的前提下，处理短时间的流量爆发。它是一个“弹性器”，在平均速率限制下提供突发能力。

6. 实现细节与考量

在实际生产环境中，实现限流远比上述代码示例复杂，需要考虑更多因素：

• 分布式限流： 当服务部署在多个实例上时，需要一个中心化的组件来管理令牌或水桶状态，例如使用 Redis 来存储限流器的状态。Redis 的 INCR 和 EXPIRE 命令、Lua 脚本在实现原子操作和复杂逻辑上非常有用。
• 时间精度： 算法依赖于精确的时间戳计算，系统时钟的偏差或调整可能影响限流效果。
• 参数动态调整： 限流参数（容量、速率）可能需要根据系统负载、业务需求动态调整，需要提供相应的配置接口。
• 多维度限流： 除了全局限流，还可能需要针对用户ID、IP地址、API路径、租户等进行多维度限流。
• 限流策略： 拒绝请求是最常见的策略，但也可以考虑排队（延迟响应）、降级（返回默认数据）、重定向等。
• 监控与告警： 实时监控限流器的状态（如拒绝率、当前水量/令牌数），并在达到阈值时发出告警。
• 高可用： 限流组件本身也需要是高可用的，避免单点故障。

例如，在分布式场景下，一个基于 Redis 的令牌桶实现伪代码可能如下：

# Lua 脚本，确保原子性
LUA_SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local fill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested_tokens = tonumber(ARGV[4])

local last_fill_time = tonumber(redis.call('HGET', key, 'last_fill_time'))
local current_tokens = tonumber(redis.call('HGET', key, 'current_tokens'))

if not last_fill_time then
    last_fill_time = now
    current_tokens = capacity
end

local time_elapsed = now - last_fill_time
local tokens_to_add = time_elapsed * fill_rate
current_tokens = math.min(capacity, current_tokens + tokens_to_add)

if current_tokens >= requested_tokens then
    current_tokens = current_tokens - requested_tokens
    redis.call('HSET', key, 'current_tokens', current_tokens)
    redis.call('HSET', key, 'last_fill_time', now)
    return 1 -- 允许
else
    redis.call('HSET', key, 'current_tokens', current_tokens) -- 即使不通过，也要更新令牌数和时间
    redis.call('HSET', key, 'last_fill_time', now)
    return 0 -- 拒绝
end
"""

# Python 客户端调用 Redis Lua 脚本
# def distributed_token_bucket_acquire(redis_client, key, capacity, fill_rate, requested_tokens=1):
#     now = time.time()
#     # 将参数传递给Lua脚本
#     result = redis_client.eval(LUA_SCRIPT, 1, key, capacity, fill_rate, now, requested_tokens)
#     return result == 1

这样的实现确保了在并发环境下，令牌的生成和消耗是原子性的，避免了竞态条件。

7. 权衡与选择

在实际应用中，选择漏桶还是令牌桶，取决于你的具体业务需求和系统瓶颈：

• 如果你的核心目标是保护后端服务，确保其负载始终平稳，且对短时突发响应不敏感，选择漏桶算法。 它能提供最稳定的服务输出速率。
• 如果你希望在保护服务的同时，允许系统在短时间内应对一定的流量冲击，提供更灵活的用户体验，选择令牌桶算法。 它能在平均速率受控的前提下，提供瞬时的高吞吐量。

很多时候，也可以将两种算法结合使用，或者采用更复杂的混合策略，以达到最佳的限流效果。例如，可以在系统的入口处使用令牌桶算法来吸收突发流量，而在核心业务逻辑或关键资源前使用漏桶算法来平滑请求，确保核心系统稳定。

尾声：精妙的平衡艺术

限流是一门精妙的平衡艺术。它不仅关乎技术实现，更关乎对业务场景的深刻理解、对系统容量的准确评估，以及对用户体验的细致考量。无论是漏桶的稳健，还是令牌桶的灵活，理解它们的内在机制与行为差异，将帮助我们更好地设计和构建高可用、高性能的分布式系统，在流量洪峰面前，依然能够从容不迫。