CC攻击（Challenge Collapsar）最棘手的地方不在于它的总量有多大，而在于它会在极短时间内释放出远超正常业务的脉冲流量。比如一个电商接口平时每秒200次请求，突然一秒钟内涌入20000次，这种短时脉冲如果不处理，后端服务直接就崩了。解决这个问题的核心思路就是把令牌桶和漏桶两种限流算法结合起来——令牌桶负责应对突发，漏桶负责平滑输出，两者叠加形成一道"先蓄后放"的双重防线。下面我把这套方案从原理到实现，从头到尾讲透。

一、为什么单独用令牌桶或者漏桶都不够

先说令牌桶。令牌桶的工作方式是系统以固定速率往桶里放令牌，每个请求来了先拿一个令牌，拿不到就拒绝。它的好处是允许一定程度的突发——因为桶里可以预先攒一些令牌。但问题在于，令牌桶只管"进"，不管"出"。如果攻击者在一秒内把桶里攒的令牌全部消耗完，后续请求虽然会被限制，但前面那一波脉冲已经打到后端了。对于CC攻击来说，这一波脉冲往往就够把服务打挂。

再说漏桶。漏桶的逻辑是请求先进一个队列（桶），然后以固定速率从桶底漏出去处理。它的优点是输出绝对平稳，不管进来多少，出去的速率恒定。但漏桶的问题是它对突发的容忍度太低——如果桶满了，多余的请求直接丢弃，而且它没有"预存"能力，无法利用系统空闲时的处理余量。

所以单独用任何一种都有缺陷。CC防护的实际场景需要的是：既能在平时攒一些处理能力应对突发，又能保证输出到后端的流量是平滑可控的。这就是令牌桶加漏桶组合方案的出发点。

二、令牌桶+漏桶组合架构的工作原理

组合方案的核心架构是这样的：请求先经过令牌桶做第一层筛选，拿到令牌的请求再进入漏桶做第二层整形，最后漏桶以恒定速率把请求放出去交给后端处理。

具体来说，令牌桶在这里承担的角色是"准入控制"。它设定一个令牌生成速率（比如每秒500个令牌）和一个桶容量上限（比如2000个令牌）。平时系统空闲时，令牌会不断积累到桶满。当CC脉冲来袭时，桶里预存的令牌可以消化掉第一波冲击，避免请求全部被直接拒绝导致误杀正常用户。但令牌桶放行的请求并不是立刻打到后端，而是先进漏桶排队。

漏桶在这里承担的角色是"流量整形"。它设定一个固定的输出速率（比如每秒300个请求）。不管令牌桶放进来多少请求，漏桶都按这个速率匀速往外放。这样即使令牌桶在突发期间放了大量请求进来，后端实际收到的流量也是平稳的，不会被瞬间冲垮。

用一个比喻来说：令牌桶像是水库的闸门，平时蓄水，洪水来了先用库存顶一阵；漏桶像是下游的灌溉渠，不管上游放多少水，到了这里都按固定流量慢慢流。两者配合，既不浪费平时的处理能力，又能扛住突发洪峰。

三、关键参数怎么设定才合理

参数设定是这套方案能不能落地的关键。设错了，要么误杀太多正常用户，要么防护形同虚设。核心参数有四个：令牌生成速率、令牌桶容量、漏桶容量、漏桶输出速率。

令牌生成速率（token_rate）：这个值应该根据你业务的正常平均QPS来定。比如你的接口正常峰值是每秒400次，那令牌生成速率可以设在400到500之间。设太低，正常高峰期也会被限流；设太高，防护效果打折扣。

令牌桶容量（token_bucket_size）：这个值决定了你能扛多大的突发。假设你的漏桶输出速率是300/s，令牌生成速率是500/s，那么桶容量设为2000意味着你可以在4秒内不生成新令牌的情况下，持续以500/s的速度放行请求。对于CC攻击常见的1-3秒脉冲，这个容量通常够用。但如果攻击者持续高压，桶很快会空，后续就靠令牌生成速率慢慢恢复。

漏桶容量（leaky_bucket_size）：这个是漏桶队列的最大长度。设太小，突发期间大量请求会被直接丢弃；设太大，队列积压会导致请求延迟飙升。一般建议设为漏桶输出速率的3到5倍，比如输出速率300/s，队列容量设1000到1500。

漏桶输出速率（leak_rate）：这个值应该根据你后端服务的实际承载能力来定。它是你真正打到后端的流量上限，必须保守。宁可限流多一点，也不能让后端过载。建议在后端压测得出的安全QPS基础上再打个七折。

四、具体实现代码示例

下面给一个Python实现的核心逻辑，生产环境需要根据实际情况做优化，但原理是完整的：

import time
import threading

class TokenBucket:
    def __init__(self, token_rate, bucket_size):
        self.token_rate = token_rate      # 每秒生成令牌数
        self.bucket_size = bucket_size    # 桶容量上限
        self.tokens = bucket_size         # 当前令牌数
        self.last_time = time.time()
        self.lock = threading.Lock()

    def acquire(self, tokens=1):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_time
            # 补充令牌
            self.tokens = min(self.bucket_size, self.tokens + elapsed * self.token_rate)
            self.last_time = now
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

class LeakyBucket:
    def __init__(self, leak_rate, bucket_size):
        self.leak_rate = leak_rate        # 每秒漏出请求数
        self.bucket_size = bucket_size    # 队列容量
        self.queue = []                   # 请求队列
        self.last_leak_time = time.time()
        self.lock = threading.Lock()

    def add_request(self, request):
        with self.lock:
            # 先执行漏出逻辑
            self._leak()
            if len(self.queue) < self.bucket_size:
                self.queue.append(request)
                return True
            return False  # 队列满，丢弃

    def _leak(self):
        now = time.time()
        elapsed = now - self.last_leak_time
        leak_count = int(elapsed * self.leak_rate)
        for _ in range(leak_count):
            if self.queue:
                self.queue.pop(0)  # 取出请求交给后端处理
        self.last_leak_time = now

class CCProtection:
    def __init__(self, token_rate, token_bucket_size, leak_rate, leaky_bucket_size):
        self.token_bucket = TokenBucket(token_rate, token_bucket_size)
        self.leaky_bucket = LeakyBucket(leak_rate, leaky_bucket_size)

    def handle_request(self, request):
        # 第一层：令牌桶准入
        if not self.token_bucket.acquire():
            return False, "rejected_by_token_bucket"
        # 第二层：漏桶整形
        if not self.leaky_bucket.add_request(request):
            return False, "rejected_by_leaky_bucket"
        return True, "accepted"

五、实际部署中需要注意的几个坑

第一，分布式环境下的令牌同步问题。如果你是多节点部署，每个节点各自维护一个令牌桶，那总的放行量会是单节点的N倍。解决办法是用Redis做集中式令牌桶，所有节点共享同一个令牌计数。Redis的INCRBY和EXPIRE命令天然适合实现分布式令牌桶。

第二，漏桶的异步处理不能阻塞。漏桶往外放请求的时候，应该是异步的，由一个独立的后台线程或者协程按速率从队列里取请求发给后端。如果在请求处理线程里同步等待漏桶放行，会严重影响吞吐。

第三，要区分正常突发和攻击脉冲。不是所有突发都是CC攻击。比如秒杀活动、爬虫集中抓取，这些也会产生脉冲。可以结合IP维度的令牌桶来做精细化控制——单个IP的令牌桶容量设小一些，全局令牌桶容量设大一些。这样单个IP的脉冲被限制，但全局不会误杀。

第四，监控和动态调整。参数不是设完就不管了。你需要实时监控令牌桶的消耗速度、漏桶的队列长度、后端的响应时间和错误率。如果发现漏桶队列长期积压，说明漏桶输出速率设低了；如果后端频繁过载，说明整体限流还不够紧。建议做一个自动调参的反馈回路。

六、进阶策略：多级防护叠加

令牌桶加漏桶只是基础防线。在实际的CC防护体系里，这套组合通常还要和其他策略叠加使用。

第一级是网络层的SYN Cookie和连接频率限制，在TCP握手阶段就过滤掉一部分恶意连接。第二级就是我们讲的令牌桶+漏桶应用层限流。第三级可以加行为分析，比如检测请求的User-Agent分布、请求间隔规律、URL访问模式，对疑似CC的IP做更激进的限流或者验证码挑战。第四级是后端的熔断降级，当限流也扛不住的时候，自动降级非核心功能，保住核心链路。

这种多级防护的好处是每一层都不需要做到极致，层层分担压力，整体系统的鲁棒性会好很多。令牌桶加漏桶在这套体系里扮演的是最核心的流量整形角色。

七、总结

CC防护的本质是在"不误杀正常用户"和"挡住恶意脉冲"之间找平衡。令牌桶给你突发容忍度，漏桶给你输出平稳度，两者结合就是一套既有弹性又有纪律的流量管控方案。关键在于参数要根据你自己的业务数据来调，不能照搬别人的数字。部署的时候注意分布式一致性、异步处理、精细化控制和动态监控，这套方案就能在大多数CC攻击场景下稳住后端服务。