2. 流量控制基础：概念、整形与算法原理

流量控制，说白了就是给网络流量装上“水龙头”。你想想看，如果没有这个水龙头，突发流量一来，后端服务直接被打垮。我在早期做支付系统时，就吃过这个亏——双十一流量峰值直接把数据库连接池打满，整个系统挂了十分钟。从那以后，我对流量控制就特别上心。

2.1 流量控制的核心概念

流量控制要解决的根本问题，是发送速率与接收能力的不匹配。它不像拥塞控制那样关注网络路径的整体状态，而是聚焦在单点——比如你的API网关、消息队列或者数据库连接池。

我习惯把流量控制分为两个层面：速率限制和流量整形。很多人把这两个概念混为一谈，其实它们有本质区别。

2.2 速率限制 vs 流量整形

维度	速率限制	流量整形
核心行为	拒绝超出速率的请求	缓存并平滑发送超出速率的请求
对用户影响	直接返回429/503	延迟处理，但最终完成
典型场景	API限流、登录防爆破	视频流、日志上报、数据同步
实现复杂度	低，计数器即可	高，需要队列和调度

举个例子：你有个API接口，每秒最多处理100个请求。速率限制的做法是——第101个请求直接返回“请求太频繁”。而流量整形会把多出来的请求先放到队列里，等下一秒再处理。嗯，这里要注意：流量整形虽然友好，但会引入延迟，实时性要求高的场景要慎用。

2.3 令牌桶算法

令牌桶是我个人最常用的算法。它的思路很简单：系统以固定速率往桶里放令牌，每个请求来的时候必须拿走一个令牌。桶有容量上限，满了令牌就丢弃。

来看一个简单的Python实现：

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 令牌生成速率（个/秒）
        self.capacity = capacity  # 桶容量
        self.tokens = capacity    # 当前令牌数
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    def try_acquire(self, count=1):
        with self.lock:
            self._refill()
            if self.tokens >= count:
                self.tokens -= count
                return True
            return False

# 使用示例：每秒10个令牌，桶容量20
bucket = TokenBucket(rate=10, capacity=20)
for i in range(30):
    if bucket.try_acquire():
        print(f"请求 {i}: 通过")
    else:
        print(f"请求 {i}: 被限")
    time.sleep(0.05)

这段代码里，_refill方法每次都会计算时间差，补上这段时间内应该生成的令牌。注意我用了锁，因为多线程环境下令牌数是共享资源。

2.4 漏桶算法

漏桶的思路正好反过来。它像一个底部有洞的水桶，水（请求）以固定速率从洞口漏出。无论你往桶里倒水多快，流出的速率始终恒定。

为什么会这样？因为漏桶强制输出速率等于漏孔大小。桶满了，新来的请求就直接丢弃。这跟令牌桶的“允许突发”正好相反。

特性	令牌桶	漏桶
输出速率	允许短时突发	严格恒定
实现方式	累加令牌	漏出请求
适用场景	API限流、突发型业务	平滑输出、流量整形
内存开销	低，只需计数器	高，需要队列

漏桶的代码实现更直观：

import time
import queue
import threading

class LeakyBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 漏出速率（个/秒）
        self.capacity = capacity  # 桶容量
        self.queue = queue.Queue(maxsize=capacity)
        self.last_leak = time.time()
        self.lock = threading.Lock()

    def try_acquire(self):
        with self.lock:
            if self.queue.qsize() < self.capacity:
                self.queue.put(time.time())
                return True
            return False

    def _leak(self):
        while True:
            time.sleep(1.0 / self.rate)
            with self.lock:
                if not self.queue.empty():
                    self.queue.get()

# 启动漏出线程
bucket = LeakyBucket(rate=5, capacity=10)
leak_thread = threading.Thread(target=bucket._leak, daemon=True)
leak_thread.start()

注意看，漏桶需要一个后台线程不断从队列里取数据，模拟“漏水”的过程。这个线程的休眠时间决定了漏出速率。

2.5 两种算法的选择策略

我个人总结了一个简单的选择原则：

• 需要允许突发流量 → 令牌桶。比如用户突然点击“刷新”按钮，短时间内发来10个请求，但长期平均不高。
• 需要严格平滑输出 → 漏桶。比如日志收集系统，后端写入数据库的速率必须恒定，否则会触发锁竞争。
• 资源有限且需要公平 → 令牌桶+加权。给不同租户分配不同的令牌生成速率。

2.6 核心知识体系

下面这张图总结了本章的核心逻辑关系：

这张图把流量控制拆成了三个层次：顶层是目标，中间层是策略，底层是具体算法。你从下往上看，就能理解不同算法怎么支撑不同的控制策略。

2.7 实际项目中的选择建议

最后，我分享几个实战中的选择建议：

• API网关限流：用令牌桶。允许用户短时间多发几个请求，体验好。
• 数据库写入限流：用漏桶。数据库对写入速率波动敏感，恒定输出最安全。
• 消息队列消费：用令牌桶+动态速率。根据队列积压情况调整令牌生成速率。
• 日志上报：用漏桶。后端存储系统通常需要恒定写入速率。