CC防护滑动窗口计数算法与内存占用优化

CC防护（Challenge Collapsar）的核心就是限制单个IP在单位时间内的请求次数，而滑动窗口计数算法是实现这一目标最常用、最高效的方案之一。传统的固定窗口算法存在"临界突刺"问题——比如窗口边界前后各发一次请求，瞬间就能绕过限制。滑动窗口通过将时间切分成更细的粒度，用一个环形数组记录每个子窗口的计数，既解决了突刺问题，又能通过内存优化把占用控制在极低水平。下面我从算法原理、实现细节、内存优化三个层面，把这套方案彻底讲透。

滑动窗口的本质是把一个大的时间窗口（比如60秒）拆分成N个小的子窗口（比如6个10秒的子窗口）。每个子窗口独立记录该时间段内的请求次数。当一个新请求到来时，系统计算当前时间所在的子窗口索引，然后向前回溯一定范围内的所有子窗口，把它们的计数加总，得到"过去60秒内的总请求数"。如果总数没超限，就允许通过并给当前子窗口计数加一；否则直接拒绝。

举个具体例子：假设我们设定60秒内最多允许100次请求，把60秒分成6个10秒的子窗口。当前时间是第53秒，落在第6个子窗口（索引5）。系统会把索引0到5这6个子窗口的计数全部加起来，判断是否超过100。如果当前是第63秒，那索引0对应的子窗口已经过期，系统只统计索引1到5以及新的索引0，实现"窗口滑动"的效果。

用环形数组（Circular Buffer）是最经典的实现方式。核心思路是用一个固定大小的数组，配合一个"当前索引指针"不断循环覆盖旧数据。下面是一个完整的Python实现示例：

import time
import threading

class SlidingWindowCounter:
    def __init__(self, max_requests, window_seconds, bucket_count):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.bucket_seconds = window_seconds / bucket_count
        self.bucket_count = bucket_count
        # 环形数组：每个元素是 [计数, 时间戳]
        self.buckets = [[0, 0] for _ in range(bucket_count)]
        self.current_index = 0
        self.lock = threading.Lock()

    def _get_current_bucket_index(self):
        now = time.time()
        # 计算当前时间属于哪个bucket
        elapsed = int(now / self.bucket_seconds)
        return elapsed % self.bucket_count

    def _reset_expired_buckets(self, current_idx):
        now = time.time()
        for i in range(self.bucket_count):
            idx = (current_idx - i) % self.bucket_count
            bucket_time = self.buckets[idx][1]
            if bucket_time > 0 and (now - bucket_time) > self.window_seconds:
                self.buckets[idx] = [0, 0]
            else:
                break  # 后面的bucket更旧，不用继续了

    def is_allowed(self):
        with self.lock:
            now = time.time()
            current_idx = self._get_current_bucket_index()
            self._reset_expired_buckets(current_idx)

            # 统计窗口内所有bucket的计数
            total = 0
            for i in range(self.bucket_count):
                idx = (current_idx - i) % self.bucket_count
                if self.buckets[idx][1] > 0 and (now - self.buckets[idx][1]) <= self.window_seconds:
                    total += self.buckets[idx][0]
                else:
                    break

            return total < self.max_requests

    def record(self):
        with self.lock:
            if self.is_allowed():
                current_idx = self._get_current_bucket_index()
                self.buckets[current_idx][0] += 1
                self.buckets[current_idx][1] = time.time()
                return True
            return False

这段代码的关键在于三点：第一，用取模运算实现环形索引；第二，每次请求时先清理过期bucket再统计；第三，用线程锁保证并发安全。在高并发场景下，锁的粒度和性能是需要重点关注的问题。

滑动窗口算法的内存开销主要来自两部分：存储每个bucket的计数和时间戳，以及为每个IP维护一套独立的窗口数据。如果用32位整数存计数、64位浮点数存时间戳，每个bucket占12字节。假设分60个bucket、同时追踪10万个IP，内存就是60×12×100000≈72MB。看起来不多，但如果bucket数量加大、IP数量到百万级，内存就会成为瓶颈。以下是几个硬核优化方向：

不需要存完整的64位时间戳。因为bucket的时间窗口是固定的（比如1秒），我们只需要记录"这个bucket是在哪个大周期被激活的"。用32位整数存"周期编号"就够了。比如以1秒为粒度，32位无符号整数可以表示约136年的周期数，完全够用。这样每个bucket的时间信息从8字节降到4字节，节省33%。

如果单窗口限流是1000次，用16位无符号整数（最大65535）就足够了。如果是更低的限制比如100次，8位就够。根据实际业务选择最小够用的类型，能显著压缩内存。在C/C++或Rust中，这种位宽选择的收益非常明显。

大多数IP的请求频率其实很低，很多bucket长期是0。不需要一开始就为每个IP分配完整的数组。可以用哈希表（HashMap）按需分配：只有当某个IP第一次触发请求时，才创建它的窗口结构。这样对于海量低频IP，内存占用可以降低一个数量级。但要注意，哈希表本身有额外开销，需要根据IP总量和活跃比例做权衡。

# 稀疏存储伪代码示例（Python字典实现）
class SparseSlidingWindow:
    def __init__(self, max_requests, window_seconds, bucket_count):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.bucket_seconds = window_seconds / bucket_count
        self.bucket_count = bucket_count
        self.ip_windows = {}  # ip -> buckets数组
        self.lock = threading.Lock()

    def get_or_create(self, ip):
        with self.lock:
            if ip not in self.ip_windows:
                self.ip_windows[ip] = [[0, 0] for _ in range(self.bucket_count)]
            return self.ip_windows[ip]

当IP数量远超内存容量时，必须有淘汰策略。最实用的是LRU（最近最少使用）：当新IP到来且内存不足时，淘汰最久没被访问的IP窗口数据。可以用一个双向链表配合哈希表实现O(1)的淘汰操作。另一种更激进的策略是"分桶淘汰"：按IP的哈希值分到不同的内存池，每个池独立淘汰，避免全局锁竞争。

如果业务允许少量误差，可以用Count-Min Sketch或HyperLogLog这类概率数据结构来近似计数。它们用固定大小的二维数组，内存占用与IP数量无关，只与精度参数有关。比如一个误差率1%、内存仅几KB的Count-Min Sketch，就能同时追踪百万级IP的请求频率。代价是计数可能略微偏高，但对CC防护来说，稍微保守一点反而是好事。

滑动窗口算法在高并发下的性能瓶颈主要是锁竞争。几个实用的优化手段：

第一，分片锁（Sharding Lock）。把IP按哈希分成多个分片，每个分片独立加锁。比如分成256个分片，锁竞争直接降低256倍。第二，无锁CAS操作。在计数更新时用原子操作（如C++的std::atomic或Java的AtomicInteger）替代互斥锁，适合计数更新频繁但冲突不极端的场景。第三，本地缓存+异步同步。先在本地内存快速判断，再异步刷到分布式存储，适合分布式部署的场景。

固定窗口实现最简单，但有临界突刺问题；令牌桶支持突发流量但实现复杂；滑动窗口在精度和复杂度之间取得了最好的平衡。对于CC防护这种需要精确限流的场景，滑动窗口是首选。但如果你的系统对突发有一定容忍度，令牌桶可能更灵活。实际工程中，很多团队会用"滑动窗口做精确计数+令牌桶做突发控制"的组合方案。

第一，时间同步很重要。如果服务器时钟不准，bucket的过期判断就会出错，建议用NTP同步。第二，bucket数量不是越多越好。bucket越多精度越高，但内存和计算开销也越大。通常6到60个bucket是性价比最高的范围。第三，要考虑IP伪造问题。CC攻击经常伪造IP，所以滑动窗口不能只看IP，还要结合设备指纹、行为特征等多维度判断。第四，分布式环境下要考虑窗口数据的一致性，可以用Redis的Sorted Set或Lua脚本实现分布式滑动窗口。

总结来说，CC防护滑动窗口计数算法的核心是用环形数组实现时间维度的精确限流，而内存优化的关键在于数据类型压缩、稀疏存储、分层淘汰和近似算法的灵活运用。在实际项目中，没有一种方案能通吃所有场景，需要根据QPS规模、IP总量、精度要求和硬件资源做综合权衡。把这些细节都处理好，你的CC防护系统才能在高并发攻击下既稳又省。