3. API限流基础概念：四大算法原理与对比

在催收触达系统中，API限流是防止上游渠道（短信、语音、邮件）被瞬间打爆、保障系统稳定性的第一道防线。本章深入剖析四种经典限流算法的核心原理、适用场景及在催收场景中的选型建议。

3.1 计数器算法（Counter Algorithm）

计数器算法是最简单直观的限流实现。它将时间划分为固定窗口（如1秒），在每个窗口内维护一个计数器，当请求到达时计数器加1；若计数器超过阈值，则拒绝后续请求；窗口结束时计数器归零。

// 伪代码示例：计数器限流
class CounterLimiter:
    def __init__(self, limit=100, window_ms=1000):
        self.limit = limit          # 窗口内最大请求数
        self.window_ms = window_ms  # 窗口大小（毫秒）
        self.counter = 0
        self.window_start = now()

    def allow(self):
        if now() - self.window_start >= self.window_ms:
            self.counter = 0
            self.window_start = now()
        self.counter += 1
        return self.counter <= self.limit

3.2 滑动窗口算法（Sliding Window Algorithm）

滑动窗口算法是对计数器算法的改进。它将时间窗口细分为多个小格（如1秒窗口分为10个100ms小格），每个小格独立计数。窗口随着时间平滑滑动，丢弃过期小格，统计当前窗口内所有小格的累计值。

// 伪代码示例：滑动窗口限流
class SlidingWindowLimiter:
    def __init__(self, limit=100, window_ms=1000, slots=10):
        self.limit = limit
        self.window_ms = window_ms
        self.slot_ms = window_ms // slots
        self.slots = [0] * slots
        self.current_slot = 0

    def allow(self):
        self._advance_window()
        total = sum(self.slots)
        if total >= self.limit:
            return False
        self.slots[self.current_slot] += 1
        return True

    def _advance_window(self):
        now_slot = (now() // self.slot_ms) % len(self.slots)
        while self.current_slot != now_slot:
            self.current_slot = (self.current_slot + 1) % len(self.slots)
            self.slots[self.current_slot] = 0

3.3 漏桶算法（Leaky Bucket Algorithm）

漏桶算法将请求视为水滴，以恒定速率从桶底漏出（处理请求），桶容量有限。当请求注入速度超过漏出速度时，桶中积水增加；若桶已满，则新请求被丢弃。漏桶强制输出为平滑的恒定速率。

// 伪代码示例：漏桶限流
class LeakyBucketLimiter:
    def __init__(self, capacity=100, leak_rate=10):
        self.capacity = capacity      # 桶容量
        self.leak_rate = leak_rate    # 每秒漏出速率
        self.water = 0                # 当前水量
        self.last_leak = now()

    def allow(self):
        self._leak()
        if self.water >= self.capacity:
            return False
        self.water += 1
        return True

    def _leak(self):
        elapsed = now() - self.last_leak
        self.water = max(0, self.water - self.leak_rate * elapsed)
        self.last_leak = now()

3.4 令牌桶算法（Token Bucket Algorithm）

令牌桶算法以恒定速率向桶中放入令牌，每个请求需消耗一个令牌才能通过。桶有最大容量限制。与漏桶不同，令牌桶允许一定程度的突发——当桶中积累足够令牌时，可一次性处理多个请求。

// 伪代码示例：令牌桶限流
class TokenBucketLimiter:
    def __init__(self, capacity=100, refill_rate=10):
        self.capacity = capacity        # 桶容量
        self.refill_rate = refill_rate  # 每秒令牌补充速率
        self.tokens = capacity          # 当前令牌数
        self.last_refill = now()

    def allow(self):
        self._refill()
        if self.tokens < 1:
            return False
        self.tokens -= 1
        return True

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

3.5 四大算法综合对比

3.6 催收场景选型建议

• 短信渠道： 推荐令牌桶算法。短信网关通常支持短时批量提交，令牌桶的突发能力可充分利用渠道吞吐量。
• 语音外呼： 推荐滑动窗口算法。语音通道对并发敏感，滑动窗口能精确控制每秒并发数，避免运营商限呼。
• 邮件发送： 推荐漏桶算法。邮件服务商通常有严格的每秒发送上限，漏桶的恒定速率特性可避免被判定为垃圾邮件。
• 日总量控制： 推荐计数器算法。实现简单，只需在Redis中维护一个Key，过期时间设为24小时即可。

3.7 本章小结

• 计数器算法： 实现最简单，但存在窗口边界突发问题，适合粗粒度限制。
• 滑动窗口算法： 通过细粒度分格消除边界突变，适合精确并发控制。
• 漏桶算法： 强制恒定输出速率，适合下游处理能力固定的场景。
• 令牌桶算法： 兼顾平滑与突发，是催收触达系统最推荐的限流方案。
• 生产建议： 采用“令牌桶 + 滑动窗口”双层架构，实现长期平滑、短期可控的限流效果。