从算法到系统设计

第四十二章：从算法到系统设计

你花了 40 章学会了数据结构与算法。现在面试官说："设计一个类似 Twitter 的系统。"你的算法知识突然不知道往哪放了。

这是因为大多数人把"算法面试"和"系统设计面试"当成两个完全不同的领域。事实并非如此——系统设计的每一个核心决策，底层都有一个数据结构或算法在支撑。一致性哈希是哈希表的分布式版本，B+Tree 索引是二叉搜索树的工程优化，布隆过滤器是集合成员查询的概率版本，消息队列的优先级调度是堆的应用。

本章做四件事：第一，展示本书前面章节的算法如何在真实系统中体现；第二，介绍分布式场景下的新算法（Raft、CRDT、DHT）；第三，与本站 Redis/Kafka/MySQL 书形成内链知识网络；第四，教你在系统设计面试中如何精确引用算法知识来加分。

Level 1 · 你需要知道的

1.1 算法在系统设计中的直接体现

每个系统设计决策的背后都有算法。下表列出最常见的映射关系：

一致性哈希（引用第 8 章）

场景： 你有 N 台缓存服务器，需要决定每个 key 存在哪台机器上。

朴素方案： server = hash(key) % N。问题：当增加或移除一台服务器时，几乎所有 key 都要重新映射——这意味着缓存几乎全部失效（cache stampede）。

一致性哈希解决了什么？ 当节点增减时，平均只有 K/N 个 key 需要重新映射（K 是总 key 数，N 是节点数）。

import hashlib
from bisect import bisect_right

class ConsistentHash:
    """一致性哈希环实现
    
    核心思想：
    1. 将哈希值空间组织成一个环（0 ~ 2^32-1）
    2. 每个节点在环上占据多个位置（虚拟节点）
    3. 每个 key 顺时针找到最近的节点
    
    虚拟节点的作用（本书第 8 章详解）：
    - 如果每个物理节点只有 1 个位置，负载分布极不均匀
    - 150-200 个虚拟节点时，标准差降到均值的 5-10%
    - 这是 Karger et al. (1997) 论文中的核心贡献
    """
    
    def __init__(self, nodes: list[str], replicas: int = 150):
        self.replicas = replicas
        self.ring = []          # (hash_value, node_name) 的有序列表
        self.hash_to_node = {}  # hash_value → node_name
        
        for node in nodes:
            self.add_node(node)
    
    def _hash(self, key: str) -> int:
        """使用 MD5 产生均匀分布的哈希值"""
        digest = hashlib.md5(key.encode()).hexdigest()
        return int(digest[:8], 16)  # 取前 32 bit
    
    def add_node(self, node: str):
        """添加物理节点 → 在环上添加 replicas 个虚拟节点"""
        for i in range(self.replicas):
            virtual_key = f"{node}#VN{i}"
            h = self._hash(virtual_key)
            self.ring.append(h)
            self.hash_to_node[h] = node
        self.ring.sort()
    
    def remove_node(self, node: str):
        """移除物理节点 → 移除其所有虚拟节点"""
        for i in range(self.replicas):
            virtual_key = f"{node}#VN{i}"
            h = self._hash(virtual_key)
            self.ring.remove(h)
            del self.hash_to_node[h]
    
    def get_node(self, key: str) -> str:
        """给定 key，找到负责的节点（顺时针第一个）
        
        算法：二分查找 → O(log(N * replicas))
        """
        if not self.ring:
            return None
        h = self._hash(key)
        # bisect_right 找到第一个 > h 的位置
        idx = bisect_right(self.ring, h) % len(self.ring)
        return self.hash_to_node[self.ring[idx]]


# 演示：节点增减时的 key 迁移量
def demo_migration():
    nodes = [f"server-{i}" for i in range(5)]
    ch = ConsistentHash(nodes)
    
    # 10000 个 key 的初始分配
    assignments_before = {}
    for i in range(10000):
        key = f"user:{i}"
        assignments_before[key] = ch.get_node(key)
    
    # 添加一个新节点
    ch.add_node("server-5")
    
    # 统计有多少 key 需要迁移
    migrated = 0
    for i in range(10000):
        key = f"user:{i}"
        if ch.get_node(key) != assignments_before[key]:
            migrated += 1
    
    # 理论值：10000 / 6 ≈ 1667（只有新节点负责的那部分需要迁移）
    print(f"迁移 key 数: {migrated} / 10000 ({migrated/100:.1f}%)")
    # 实际输出约 16-20%，接近理论值 1/6 ≈ 16.7%

布隆过滤器（引用第 9 章）

场景： 爬虫系统有 10 亿个已爬取的 URL，每次发现新 URL 需要判断是否已爬过。用 HashSet 存 10 亿个 URL 需要约 60GB 内存。布隆过滤器只需要约 1.2GB（以 1% 误判率计算）。

import mmh3  # MurmurHash3，非加密哈希，速度快
from bitarray import bitarray

class BloomFilter:
    """布隆过滤器实现
    
    数学基础（本书第 9 章详解）：
    - m = -(n * ln(p)) / (ln(2))²  最优 bit 数
    - k = (m/n) * ln(2)            最优哈希函数数
    - 其中 n = 预期元素数，p = 目标误判率
    
    空间优势：HashSet 存 n 个元素需要 O(n * 元素大小)
    布隆过滤器只需 O(n) bits，与元素大小无关
    
    代价：
    - 有误判（False Positive）：说"存在"可能是错的
    - 无漏判（No False Negative）：说"不存在"一定是对的
    - 不支持删除（Counting Bloom Filter 可以，但空间翻倍）
    """
    
    def __init__(self, expected_items: int, fp_rate: float = 0.01):
        import math
        self.size = int(-(expected_items * math.log(fp_rate)) / (math.log(2) ** 2))
        self.num_hashes = int((self.size / expected_items) * math.log(2))
        self.bit_array = bitarray(self.size)
        self.bit_array.setall(0)
        
        # 空间计算示例：
        # n=10亿, p=1% → m≈95.8亿 bits ≈ 1.2GB, k=7
        # 对比 HashSet：每个 URL 平均 50 bytes → 50GB
    
    def add(self, item: str):
        """添加元素：对每个哈希函数，将对应位设为 1"""
        for i in range(self.num_hashes):
            idx = mmh3.hash(item, i) % self.size
            self.bit_array[idx] = 1
    
    def contains(self, item: str) -> bool:
        """查询元素：所有哈希位都为 1 → 可能存在；任一为 0 → 一定不存在"""
        for i in range(self.num_hashes):
            idx = mmh3.hash(item, i) % self.size
            if not self.bit_array[idx]:
                return False  # 一定不存在
        return True  # 可能存在（有 fp_rate 概率误判）


# 在系统设计中如何使用
class WebCrawler:
    """带布隆过滤器的爬虫去重"""
    
    def __init__(self, expected_urls: int = 1_000_000_000):
        self.bloom = BloomFilter(expected_urls, fp_rate=0.001)
        # 1% 误判意味着 1000 万个 URL 可能被错误跳过
        # 对爬虫来说可以接受：下次爬取周期会补上
    
    def should_crawl(self, url: str) -> bool:
        """决定是否爬取该 URL"""
        if self.bloom.contains(url):
            return False  # 可能已爬过（或误判），跳过
        return True
    
    def mark_crawled(self, url: str):
        """标记 URL 为已爬取"""
        self.bloom.add(url)

B+Tree 索引（引用第 20 章）

场景： MySQL 执行 SELECT * FROM users WHERE age BETWEEN 25 AND 35，如何高效找到所有满足条件的行？

class BPlusTreeNode:
    """B+Tree 节点 — 简化实现展示核心思想
    
    为什么数据库选择 B+Tree 而非红黑树/AVL？
    1. 磁盘友好：每个节点是一个磁盘页（通常 16KB），减少 I/O 次数
    2. 范围查询高效：叶子节点形成有序链表，范围查询只需顺序扫描
    3. 高扇出（fanout）：内部节点只存 key，不存数据，能容纳更多 key
       → 树高极低（3-4 层即可索引数十亿行）
    
    与二叉搜索树的对比：
    - BST：高度 O(log₂ n) → 10 亿行需要 30 次比较 → 30 次磁盘 I/O
    - B+Tree（阶=1000）：高度 O(log₁₀₀₀ n) → 10 亿行只需 3 次 I/O!
    """
    
    def __init__(self, is_leaf: bool = False, order: int = 4):
        self.is_leaf = is_leaf
        self.keys = []        # 排序的 key 列表
        self.children = []    # 内部节点：子节点指针；叶子节点：数据/记录指针
        self.next_leaf = None  # 叶子节点的链表指针（用于范围查询）
        self.order = order    # B+Tree 的阶（实际中通常 100-1000）


def range_query_explanation():
    """B+Tree 范围查询流程（概念说明）
    
    SQL: SELECT * FROM users WHERE age BETWEEN 25 AND 35
    
    步骤：
    1. 从根节点出发，用二分查找定位 age=25 所在的叶子节点  → O(log_B n) 次 I/O
    2. 在叶子节点中找到第一个 >= 25 的位置
    3. 沿叶子链表顺序扫描，直到 age > 35 停止           → O(结果集大小 / 页大小) 次 I/O
    
    关键优势：
    - 定位开销极低（3-4 次 I/O）
    - 范围扫描是顺序 I/O（SSD 顺序读 3GB/s vs 随机读 100MB/s）
    """
    pass

1.2 从单机算法到分布式算法

当你的系统从单机扩展到多机，单机上的"确定性"消失了：

这些变化催生了一系列分布式算法。以下是你在系统设计面试中最可能遇到的三个：

Raft 共识算法（简要介绍）：

class RaftState:
    """Raft 共识算法的状态机模型
    
    解决的问题：多个节点如何对"日志的顺序"达成一致？
    
    学术来源：Ongaro & Ousterhout, "In Search of an Understandable Consensus Algorithm"
    (USENIX ATC 2014) — 专门设计为 Paxos 的可理解版本
    
    核心角色：
    - Leader：接收客户端请求，复制日志到 Follower
    - Follower：被动接收日志
    - Candidate：在选举期间竞争成为 Leader
    
    关键不变量：
    - 每个 term 最多一个 Leader
    - Leader 的日志一定包含所有已提交的条目
    - 多数派同意即可提交
    """
    
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"
    
    def __init__(self, node_id: int, total_nodes: int):
        self.node_id = node_id
        self.state = self.FOLLOWER
        self.current_term = 0
        self.voted_for = None
        self.log = []  # [(term, command), ...]
        self.commit_index = 0
        
        # Leader 特有状态
        self.next_index = {}   # 对每个 follower，下一个要发送的日志索引
        self.match_index = {}  # 对每个 follower，已确认复制的最高日志索引
        
        self.majority = total_nodes // 2 + 1
    
    def request_vote(self, candidate_term: int, candidate_id: int,
                     last_log_index: int, last_log_term: int) -> bool:
        """投票 RPC：决定是否给候选人投票
        
        投票条件：
        1. candidate_term >= 自己的 term
        2. 本 term 还没投过票（或者已投给该候选人）
        3. 候选人的日志至少和自己一样新（先比 term，再比 index）
        """
        if candidate_term < self.current_term:
            return False
        
        if candidate_term > self.current_term:
            self.current_term = candidate_term
            self.state = self.FOLLOWER
            self.voted_for = None
        
        if self.voted_for is not None and self.voted_for != candidate_id:
            return False
        
        # 日志新鲜度比较（关键！保证 Leader Completeness）
        my_last_term = self.log[-1][0] if self.log else 0
        my_last_index = len(self.log)
        
        log_is_up_to_date = (
            candidate_term > my_last_term or
            (candidate_term == my_last_term and last_log_index >= my_last_index)
        )
        
        if log_is_up_to_date:
            self.voted_for = candidate_id
            return True
        return False
    
    def append_entries(self, leader_term: int, entries: list,
                       prev_log_index: int, prev_log_term: int,
                       leader_commit: int) -> bool:
        """日志追加 RPC：Leader 向 Follower 复制日志
        
        Follower 检查：
        1. Leader 的 term >= 自己的 term
        2. 在 prev_log_index 位置，term 必须匹配
           → 不匹配说明日志有分歧，需要回退
        """
        if leader_term < self.current_term:
            return False
        
        self.current_term = leader_term
        self.state = self.FOLLOWER
        
        # 一致性检查
        if prev_log_index > 0:
            if len(self.log) < prev_log_index:
                return False  # 日志太短
            if self.log[prev_log_index - 1][0] != prev_log_term:
                # 日志冲突：删除该位置及之后的所有条目
                self.log = self.log[:prev_log_index - 1]
                return False
        
        # 追加新条目
        self.log.extend(entries)
        
        # 更新 commit_index
        if leader_commit > self.commit_index:
            self.commit_index = min(leader_commit, len(self.log))
        
        return True

1.3 系统设计面试中的算法引用模板

面试时如何精确引用算法知识：

# 面试官："如何设计一个分布式缓存系统？"
# 
# 你的回答框架：
SYSTEM_DESIGN_WITH_ALGORITHMS = """
1. 数据分片（Sharding）
   → 使用一致性哈希（Karger et al., 1997）
   → 每个物理节点 150 个虚拟节点保证均匀分布
   → 节点增减时只需迁移 1/N 的数据

2. 缓存淘汰
   → LRU 策略：哈希表 + 双向链表，O(1) get/put
   → 如果需要考虑频率：LFU 或 W-TinyLFU（Caffeine 的实现）

3. 热点检测
   → Count-Min Sketch（概率数据结构，第 9 章）
   → 用 O(1) 空间估计每个 key 的访问频率

4. 缓存穿透防护
   → 布隆过滤器判断 key 是否可能存在于数据库中
   → 不存在的 key 直接返回，不查数据库

5. 一致性保证
   → 如果要求强一致：Raft 复制日志
   → 如果允许最终一致：gossip 协议 + 向量时钟
"""

1.4 常见错误

错误 1：混淆"算法"和"数据结构"的应用层次

# ❌ 错误：在系统设计中说"用红黑树做索引"
# 红黑树是内存数据结构，磁盘上应该用 B+Tree

# ❌ 错误：在系统设计中说"用 DFS 做服务发现"
# DFS 是图遍历算法，服务发现应该用 gossip 或注册中心

# ✅ 正确的映射：
CORRECT_MAPPINGS = {
    "内存索引": "红黑树、跳表（Redis）",
    "磁盘索引": "B+Tree、LSM-Tree",
    "内存缓存淘汰": "LRU（哈希+双链表）",
    "分布式路由": "一致性哈希",
    "近似计数": "HyperLogLog",
    "近似判存": "布隆过滤器",
    "分布式共识": "Raft / Paxos",
}

错误 2：过度设计小规模系统

当面试官说"用户量 10 万"时，你不需要一致性哈希——单机 Redis 就够了。算法和数据结构的选择必须匹配规模。

Level 2 · 它是怎么运行的

2.1 分布式算法深度解析

Raft 的工程实现细节

Raft 被 etcd（Kubernetes 的底层存储）、CockroachDB、TiKV（TiDB 的存储引擎）等系统采用。下面分析其核心机制：

选举机制的数学保证：

import random

class RaftElection:
    """Raft 选举的关键设计决策
    
    为什么使用随机超时？
    - 如果所有节点同时开始选举，可能永远无法选出 Leader（活锁）
    - 随机超时 [150ms, 300ms] 确保某个节点大概率先超时发起选举
    - 数学分析：如果超时范围远大于网络 RTT，冲突概率趋近于 0
    
    超时范围的选择（Ongaro 论文建议）：
    - 下界 > 平均网络 RTT（避免频繁选举）
    - 上界 < 可接受的不可用时间
    - 范围 = 上界 - 下界 >> 平均 RTT（减少冲突）
    """
    
    def __init__(self, heartbeat_interval: int = 50):
        self.heartbeat_interval = heartbeat_interval  # ms
        # 选举超时 = [2*heartbeat, 4*heartbeat] — 经验值
        self.election_timeout_range = (
            heartbeat_interval * 3,   # 150ms
            heartbeat_interval * 6    # 300ms
        )
    
    def random_election_timeout(self) -> int:
        """返回随机选举超时时间（ms）"""
        return random.randint(*self.election_timeout_range)
    
    def simulate_election(self, num_nodes: int = 5, 
                          network_delay_ms: int = 5) -> dict:
        """模拟一次选举过程
        
        最佳情况：1 RTT 完成选举（150 + 5 = 155ms）
        最坏情况：多次冲突（极少见，期望 < 2 次）
        """
        # 每个节点的超时时间
        timeouts = [self.random_election_timeout() for _ in range(num_nodes)]
        
        # 第一个超时的节点发起选举
        first_candidate = min(range(num_nodes), key=lambda i: timeouts[i])
        election_start = timeouts[first_candidate]
        
        # 投票到达时间 = 选举开始 + 网络延迟
        vote_collection_time = election_start + network_delay_ms
        
        # 如果没有其他节点同时超时（间隔 > RTT），选举成功
        second_timeout = sorted(timeouts)[1]
        conflict = (second_timeout - election_start) < network_delay_ms
        
        return {
            "winner": first_candidate,
            "time_ms": vote_collection_time,
            "conflict": conflict,
            "rounds": 2 if conflict else 1,
        }

CRDT：无冲突复制数据类型

class GCounter:
    """G-Counter（Grow-only Counter）— 最简单的 CRDT
    
    学术来源：Shapiro et al., "A comprehensive study of Convergent and 
    Commutative Replicated Data Types" (INRIA Technical Report, 2011)
    
    核心思想：
    - 每个节点维护自己的计数器
    - 全局值 = 所有节点计数器之和
    - 合并操作 = 取每个节点计数器的 max
    
    为什么需要 CRDT？
    - 传统分布式系统：要么用锁（性能差），要么用共识（延迟高）
    - CRDT：无锁、无共识、最终一致 — 适用于计数器、集合等简单数据类型
    - 代价：只能支持特定操作（单调递增/递减/添加/删除），不能支持任意操作
    """
    
    def __init__(self, node_id: str, num_nodes: int):
        self.node_id = node_id
        self.counters = {}  # node_id → count
    
    def increment(self, amount: int = 1):
        """本地递增"""
        self.counters[self.node_id] = self.counters.get(self.node_id, 0) + amount
    
    def value(self) -> int:
        """全局值 = 所有节点计数之和"""
        return sum(self.counters.values())
    
    def merge(self, other: 'GCounter'):
        """合并操作：取每个节点的 max
        
        关键性质（为什么不会冲突）：
        - 交换律：merge(A, B) == merge(B, A)
        - 结合律：merge(merge(A, B), C) == merge(A, merge(B, C))
        - 幂等性：merge(A, A) == A
        
        这三个性质保证了无论消息以什么顺序到达，最终状态一致
        """
        for node_id, count in other.counters.items():
            self.counters[node_id] = max(
                self.counters.get(node_id, 0),
                count
            )


class PNCounter:
    """PN-Counter（支持递增和递减）
    
    思路：用两个 G-Counter，一个记录增，一个记录减
    value = P.value() - N.value()
    """
    
    def __init__(self, node_id: str, num_nodes: int):
        self.p = GCounter(node_id, num_nodes)  # positive
        self.n = GCounter(node_id, num_nodes)  # negative
    
    def increment(self, amount: int = 1):
        self.p.increment(amount)
    
    def decrement(self, amount: int = 1):
        self.n.increment(amount)
    
    def value(self) -> int:
        return self.p.value() - self.n.value()
    
    def merge(self, other: 'PNCounter'):
        self.p.merge(other.p)
        self.n.merge(other.n)


class LWWRegister:
    """Last-Writer-Wins Register — 基于时间戳的 CRDT
    
    应用场景：分布式数据库中单个字段的并发更新
    - 多个节点同时修改同一字段 → 保留时间戳最大的那个
    
    局限性：
    - 依赖时钟同步（NTP 精度通常 1-10ms）
    - 如果两次写入间隔 < 时钟误差，结果不确定
    - 解决方案：Hybrid Logical Clock (HLC) — Kulkarni et al. (2014)
    """
    
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.value = None
        self.timestamp = 0  # Lamport timestamp 或物理时间
    
    def write(self, value, timestamp: int):
        if timestamp > self.timestamp:
            self.value = value
            self.timestamp = timestamp
    
    def merge(self, other: 'LWWRegister'):
        if other.timestamp > self.timestamp:
            self.value = other.value
            self.timestamp = other.timestamp

分布式哈希表（DHT）

class KademliaNode:
    """Kademlia DHT — P2P 网络中的分布式哈希表
    
    学术来源：Maymounkov & Mazières, "Kademlia: A Peer-to-peer Information 
    System Based on the XOR Metric" (IPTPS 2002)
    
    被 BitTorrent、IPFS、Ethereum 等系统采用
    
    核心思想：
    1. 每个节点有一个 160-bit ID
    2. 距离 = XOR（异或）— 满足三角不等式，但不需要物理距离
    3. 每个节点维护 160 个"桶"（k-bucket），第 i 个桶存储距离在 [2^i, 2^(i+1)) 的节点
    4. 查找操作：每步将距离减半 → O(log n) 步找到目标
    
    与一致性哈希的对比：
    - 一致性哈希：需要中心化的"哈希环"信息 → 适合数据中心内部
    - DHT：完全去中心化，每个节点只需知道 O(log n) 个其他节点 → 适合 P2P
    """
    
    def __init__(self, node_id: int, k: int = 20):
        self.node_id = node_id
        self.k = k  # 每个桶的最大容量
        self.buckets = [[] for _ in range(160)]  # 160 个 k-bucket
        self.data = {}  # 本节点存储的 key-value
    
    def distance(self, id1: int, id2: int) -> int:
        """XOR 距离"""
        return id1 ^ id2
    
    def bucket_index(self, other_id: int) -> int:
        """确定 other_id 属于哪个桶（即 XOR 距离的最高位位置）"""
        dist = self.distance(self.node_id, other_id)
        if dist == 0:
            return 0
        return dist.bit_length() - 1
    
    def find_closest_nodes(self, target_id: int, count: int = 20) -> list:
        """找到路由表中距离 target_id 最近的 count 个节点
        
        这是 Kademlia 查找的核心操作：
        每次调用，返回的节点都比上一次更接近目标
        → 保证 O(log n) 步收敛
        """
        all_nodes = []
        for bucket in self.buckets:
            all_nodes.extend(bucket)
        
        # 按 XOR 距离排序
        all_nodes.sort(key=lambda n: self.distance(n, target_id))
        return all_nodes[:count]
    
    def iterative_find_value(self, key: int) -> str:
        """迭代查找过程（简化）
        
        实际步骤：
        1. 从自己的路由表中找 α 个最近节点
        2. 并行发送 FIND_VALUE RPC
        3. 如果返回值 → 结束
        4. 如果返回更近的节点 → 更新候选列表，继续查找
        5. 重复直到找到值或候选列表不再更新
        
        复杂度：O(log n) 轮 RPC，每轮 α 个并行请求
        在 100 万节点的网络中，约 20 轮 × α=3 = 60 次 RPC 即可找到任何 key
        """
        pass

2.2 算法与系统组件的深层对应

LSM-Tree vs B+Tree：写优化 vs 读优化

class LSMTreeConcept:
    """LSM-Tree (Log-Structured Merge Tree) 概念说明
    
    学术来源：O'Neil et al., "The Log-Structured Merge-Tree (LSM-Tree)" (1996)
    
    被 LevelDB、RocksDB、Cassandra、HBase 采用
    
    核心思想：
    1. 写入先进内存（MemTable，通常是跳表或红黑树）
    2. MemTable 满后刷到磁盘成为 SSTable（Sorted String Table）
    3. 后台周期性合并（Compaction）多个 SSTable
    
    与 B+Tree 的对比：
    ┌──────────┬───────────────────┬───────────────────┐
    │          │ B+Tree            │ LSM-Tree          │
    ├──────────┼───────────────────┼───────────────────┤
    │ 写放大   │ 高（随机写磁盘）    │ 低（顺序写）      │
    │ 读放大   │ 低（1次定位）      │ 高（可能查多层）   │
    │ 空间放大  │ 低               │ 中（多版本共存）    │
    │ 适用场景  │ 读多写少（MySQL） │ 写多读少（日志系统）│
    └──────────┴───────────────────┴───────────────────┘
    """
    
    def __init__(self, memtable_size: int = 4 * 1024 * 1024):  # 4MB
        self.memtable = {}  # 模拟跳表/红黑树
        self.memtable_size = memtable_size
        self.sstables = []  # 磁盘上的有序文件列表，新的在前
    
    def put(self, key: str, value: str):
        """写入：先写 WAL（略），再写 MemTable
        
        时间复杂度：O(log n) 写入跳表
        磁盘 I/O：只有 WAL 的顺序追加写（1 次）
        """
        self.memtable[key] = value
        if len(self.memtable) >= self.memtable_size:
            self._flush()
    
    def get(self, key: str) -> str:
        """读取：先查 MemTable → 再从新到旧查 SSTable
        
        最坏情况：需要查所有层级 → 用布隆过滤器跳过不包含该 key 的 SSTable
        """
        if key in self.memtable:
            return self.memtable[key]
        
        for sstable in self.sstables:  # 从新到旧
            # 实际实现中：先查布隆过滤器，再二分查找
            if key in sstable:
                return sstable[key]
        
        return None  # key 不存在
    
    def _flush(self):
        """将 MemTable 刷到磁盘成为新的 SSTable"""
        sstable = dict(sorted(self.memtable.items()))  # 排序
        self.sstables.insert(0, sstable)  # 新的在前
        self.memtable = {}
    
    def _compact(self):
        """合并多个 SSTable（简化版 — 实际有 Leveled/Sized-Tiered 策略）
        
        为什么需要 Compaction？
        1. 减少读放大（层数减少）
        2. 回收已删除/覆盖的数据空间
        3. 保持有序性
        
        代价：写放大（数据被读出再写入），但都是顺序 I/O
        """
        if len(self.sstables) >= 4:  # 简单触发条件
            merged = {}
            for sstable in reversed(self.sstables):  # 旧的先合，新的覆盖
                merged.update(sstable)
            self.sstables = [dict(sorted(merged.items()))]

跳表在 Redis 中的应用

import random

class SkipListNode:
    def __init__(self, key: float, value: str, level: int):
        self.key = key
        self.value = value
        self.forward = [None] * (level + 1)  # 每层的 next 指针
        self.span = [0] * (level + 1)        # 每层跨越的节点数（用于 rank）

class SkipList:
    """跳表 — Redis ZSET 的底层实现
    
    为什么 Redis 选择跳表而非红黑树？（antirez 本人的解释）
    1. 实现简单：跳表 100 行代码 vs 红黑树 300+ 行
    2. 范围操作自然高效：ZRANGEBYSCORE 只需找到起点，然后顺序遍历
    3. 内存开销可控：平均每个节点 1.33 个指针（p=1/4 时）
    4. 与红黑树性能相当：O(log n) 查找/插入/删除
    
    在系统设计面试中的说法：
    "Redis 的排行榜功能用跳表实现。跳表的期望高度是 O(log n)，
    查找和范围查询都是 O(log n + k)。相比红黑树，跳表的优势在于
    范围查询不需要中序遍历，直接沿底层链表顺序扫描。"
    """
    
    MAX_LEVEL = 32
    P = 0.25  # 晋升概率（Redis 使用 1/4）
    
    def __init__(self):
        self.header = SkipListNode(float('-inf'), '', self.MAX_LEVEL)
        self.level = 0
        self.length = 0
    
    def _random_level(self) -> int:
        """随机决定新节点的层数
        
        几何分布：P(level=k) = p^(k-1) * (1-p)
        期望层数 = 1/(1-p) = 1.33（当 p=0.25）
        """
        level = 0
        while random.random() < self.P and level < self.MAX_LEVEL:
            level += 1
        return level
    
    def insert(self, key: float, value: str):
        """插入操作 — O(log n) 期望时间
        
        步骤：
        1. 从最高层开始，找到每层的插入位置（update 数组）
        2. 随机生成层数
        3. 在每层插入节点
        """
        update = [None] * (self.MAX_LEVEL + 1)
        current = self.header
        
        # 从高层到低层查找插入位置
        for i in range(self.level, -1, -1):
            while current.forward[i] and current.forward[i].key < key:
                current = current.forward[i]
            update[i] = current
        
        # 生成随机层数
        new_level = self._random_level()
        if new_level > self.level:
            for i in range(self.level + 1, new_level + 1):
                update[i] = self.header
            self.level = new_level
        
        # 创建新节点并插入每层
        new_node = SkipListNode(key, value, new_level)
        for i in range(new_level + 1):
            new_node.forward[i] = update[i].forward[i]
            update[i].forward[i] = new_node
        
        self.length += 1
    
    def range_by_score(self, min_score: float, max_score: float) -> list:
        """范围查询 — O(log n + k)，k 为结果集大小
        
        这就是 Redis ZRANGEBYSCORE 的核心逻辑
        """
        # 找到第一个 >= min_score 的节点
        current = self.header
        for i in range(self.level, -1, -1):
            while current.forward[i] and current.forward[i].key < min_score:
                current = current.forward[i]
        
        current = current.forward[0]  # 移到第一个 >= min_score 的节点
        
        # 沿底层链表扫描
        result = []
        while current and current.key <= max_score:
            result.append((current.key, current.value))
            current = current.forward[0]
        
        return result

2.3 与本站其他书的知识串联

本站（dev.yiteai.com）的技术书籍形成了一个知识网络。以下是本章内容与其他书的关联：

# 示例：Redis 中的布隆过滤器使用
"""
# Redis 命令（对应本章 BloomFilter 类的操作）
BF.RESERVE crawler_urls 0.001 1000000000  # 创建：10亿容量，0.1%误判率
BF.ADD crawler_urls "https://example.com/page1"  # 添加
BF.EXISTS crawler_urls "https://example.com/page1"  # 查询 → 1 (可能存在)
BF.EXISTS crawler_urls "https://example.com/unknown" # 查询 → 0 (一定不存在)

# 内存占用：约 1.44GB（对比 SET 存 10 亿 URL ≈ 60GB）
"""

# 示例：MySQL 索引查询对应的算法
"""
-- SQL 查询
EXPLAIN SELECT * FROM orders WHERE user_id = 12345 AND created_at > '2024-01-01';

-- 底层 B+Tree 操作（对应本章 BPlusTreeNode）：
-- 1. 在 (user_id, created_at) 联合索引的 B+Tree 上
-- 2. 定位 user_id=12345 的第一个叶子节点   → O(log_B n) 次 I/O
-- 3. 在叶子链表上扫描 created_at > '2024-01-01' → O(k/B) 次 I/O
-- 4. 回表（如果不是覆盖索引）读取完整行

-- 这就是为什么联合索引的列顺序很重要：
-- (user_id, created_at) → 先定位 user，再范围扫描 time ✓
-- (created_at, user_id) → 扫描所有 2024 年数据再过滤 user ✗
"""

2.4 消息队列中的数据结构

class DelayQueue:
    """延迟消息队列 — 基于最小堆
    
    应用场景：
    - 订单 30 分钟未支付自动取消
    - 定时推送通知
    - 重试退避（1s, 2s, 4s, 8s...）
    
    实现思路：
    - 最小堆按"到期时间"排序
    - 消费者循环检查堆顶：如果到期则取出执行
    
    在 Kafka 中的实际做法（本站 Kafka 书详解）：
    - Kafka 本身不支持延迟消息
    - 常见方案：多个延迟 topic（delay_1s, delay_5s, delay_30m...）
    - 或者用时间轮（Hierarchical Timing Wheel）实现
    """
    
    import heapq
    import time
    
    def __init__(self):
        self.heap = []  # (execute_at_timestamp, message_id, message)
        self.counter = 0  # 用于打破时间戳相同时的比较
    
    def put(self, message: str, delay_seconds: float):
        """添加延迟消息"""
        import time
        execute_at = time.time() + delay_seconds
        import heapq
        heapq.heappush(self.heap, (execute_at, self.counter, message))
        self.counter += 1
    
    def poll(self) -> str:
        """取出已到期的消息（如果有）"""
        import time, heapq
        if not self.heap:
            return None
        if self.heap[0][0] <= time.time():
            _, _, message = heapq.heappop(self.heap)
            return message
        return None  # 最早的消息还没到期


class TimingWheel:
    """时间轮 — Kafka/Netty 的定时器实现
    
    学术来源：Varghese & Lauck, "Hashed and Hierarchical Timing Wheels" (1996)
    
    为什么不用堆？
    - 堆的插入和删除是 O(log n)
    - 时间轮的插入和删除是 O(1)
    - 当有数百万个定时任务时，差异巨大
    
    核心思想（类比时钟）：
    - 秒针：60 格，每格 1 秒
    - 分针：60 格，每格 1 分钟
    - 时针：24 格，每格 1 小时
    
    任务按过期时间放入对应格子，"指针"每 tick 前进一格
    当高层格子到达时，任务下沉到低层格子（类似时钟进位的逆过程）
    """
    
    def __init__(self, tick_ms: int = 1, wheel_size: int = 60):
        self.tick_ms = tick_ms
        self.wheel_size = wheel_size
        self.buckets = [[] for _ in range(wheel_size)]
        self.current_tick = 0
        self.overflow_wheel = None  # 更高层的时间轮（延迟加载）
    
    def add_task(self, task_id: str, delay_ticks: int):
        """添加定时任务 — O(1)"""
        if delay_ticks < self.wheel_size:
            # 放入当前层
            bucket_idx = (self.current_tick + delay_ticks) % self.wheel_size
            self.buckets[bucket_idx].append((task_id, delay_ticks))
        else:
            # 溢出到更高层
            if self.overflow_wheel is None:
                self.overflow_wheel = TimingWheel(
                    self.tick_ms * self.wheel_size,
                    self.wheel_size
                )
            self.overflow_wheel.add_task(task_id, delay_ticks // self.wheel_size)
    
    def advance(self):
        """前进一个 tick — O(当前格子中的任务数)"""
        self.current_tick = (self.current_tick + 1) % self.wheel_size
        expired_tasks = self.buckets[self.current_tick]
        self.buckets[self.current_tick] = []
        return expired_tasks

Level 3 · 规范怎么定义的

3.1 CAP 定理与算法选择

CAP 定理（Brewer, 2000; Gilbert & Lynch 证明, 2002）指出分布式系统不可能同时满足三个性质：

• Consistency（一致性）：所有节点在同一时刻看到相同数据
• Availability（可用性）：每个请求都能收到非错误响应
• Partition tolerance（分区容忍性）：网络分区时系统继续运行

由于网络分区不可避免（P 必须选），实际选择是 CP vs AP：

CAP_ALGORITHM_MAPPING = {
    "CP (一致性优先)": {
        "算法": "Raft / Paxos / ZAB",
        "特征": "网络分区时少数派停止服务",
        "系统": ["etcd", "ZooKeeper", "CockroachDB", "Spanner"],
        "适用场景": "金融交易、分布式锁、配置管理",
        "本书关联": "Raft 实现（本章 Level 1）",
    },
    "AP (可用性优先)": {
        "算法": "Gossip / CRDT / Last-Writer-Wins",
        "特征": "网络分区时所有节点继续服务，但数据可能不一致",
        "系统": ["Cassandra", "DynamoDB", "Redis Cluster (部分场景)"],
        "适用场景": "社交媒体、购物车、DNS",
        "本书关联": "CRDT 实现（本章 Level 2）",
    },
}

PACELC 扩展（Abadi, 2012）：在没有分区（Partition-free）时，还有 Latency vs Consistency 的权衡：

3.2 系统设计面试的评分维度

根据 Meta 系统设计面试官的公开分享（2023 年 InfoQ 演讲）和 Designing Data-Intensive Applications（Kleppmann, 2017）的框架：

Deep Dive 中的算法引用示例：

# 面试官："排行榜如何实时更新？"
# 
# 你的回答：
"""
使用 Redis 的 ZSET（Sorted Set）。

底层实现是跳表 + 哈希表的组合：
- 跳表：维护分数排序，支持 O(log n) 的范围查询
- 哈希表：O(1) 的单 key 查找

操作复杂度：
- ZADD（更新分数）：O(log n)
- ZRANK（查排名）：O(log n) — 通过 span 累加实现
- ZRANGEBYSCORE（取 top-k）：O(log n + k)

为什么不用 MySQL ORDER BY？
- 每次查询需要全表排序或索引扫描
- 高并发下（如直播打赏排行），MySQL 承受不住
- Redis 跳表的更新是原子的，天然支持并发

容量估算：
- 1000 万用户排行榜 → ZSET 大小约 200MB
- 如果排行榜按游戏分区，每个分区 10 万人 → 单个 ZSET 约 2MB
"""

3.3 分布式系统中的经典算法定理

3.4 一致性模型谱系

从强到弱的一致性模型，及其对应的实现算法：

CONSISTENCY_SPECTRUM = {
    "Linearizability（线性一致性）": {
        "保证": "操作看起来像在单个时间点原子执行",
        "算法": "Raft/Paxos + 读写都走 Leader",
        "代价": "延迟高（每次写需要多数派确认）",
        "系统": "Spanner, CockroachDB",
    },
    "Sequential Consistency（顺序一致性）": {
        "保证": "所有操作有一个全局顺序，每个进程的操作保持原始顺序",
        "算法": "全序广播（Total Order Broadcast）",
        "代价": "中等延迟",
        "系统": "ZooKeeper (写操作)",
    },
    "Causal Consistency（因果一致性）": {
        "保证": "因果相关的操作按顺序可见，并发操作可以乱序",
        "算法": "向量时钟 / 版本向量",
        "代价": "较低延迟，允许部分并发",
        "系统": "MongoDB (w:majority + causal sessions)",
    },
    "Eventual Consistency（最终一致性）": {
        "保证": "如果没有新写入，最终所有副本收敛到相同值",
        "算法": "Gossip + CRDT / Last-Writer-Wins",
        "代价": "最低延迟，但可能读到旧数据",
        "系统": "DynamoDB, Cassandra, DNS",
    },
}

3.5 系统设计中的"数量级估算"方法

这是系统设计面试的必备技能——它直接决定你选择什么算法和数据结构：

# Jeff Dean 的"每个程序员都应该知道的延迟数字"（2012，持续更新）
LATENCY_NUMBERS = {
    "L1 cache reference":           0.5,    # ns
    "L2 cache reference":           7,      # ns
    "Main memory reference":        100,    # ns
    "SSD random read":              150_000, # ns = 150 µs
    "HDD random read":              10_000_000, # ns = 10 ms
    "Network round trip (同机房)":   500_000, # ns = 0.5 ms
    "Network round trip (跨城市)":   50_000_000, # ns = 50 ms
}

def estimate_scale(description: str) -> dict:
    """系统设计的容量估算模板
    
    面试中边算边说：
    'DAU 1000 万，每人每天 10 次请求 → QPS = 10M * 10 / 86400 ≈ 1200'
    '峰值按 3 倍估算 → 3600 QPS → 单机 Redis 轻松搞定'
    """
    # Twitter-like 系统的典型估算
    estimation = {
        "DAU": 10_000_000,
        "每人每天操作": 10,
        "总日请求": 100_000_000,
        "平均 QPS": 100_000_000 / 86400,  # ≈ 1157
        "峰值 QPS": 1157 * 3,  # ≈ 3472
        "每条推文大小": 280,  # bytes (纯文本)
        "每天新推文": 10_000_000,  # 假设 10% 用户发推
        "每天数据增量": 10_000_000 * 280,  # ≈ 2.8 GB
        "一年数据": 2.8 * 365,  # ≈ 1 TB
    }
    
    # 基于估算选择技术方案
    if estimation["峰值 QPS"] < 10000:
        estimation["缓存方案"] = "单机 Redis 足够（10万+ QPS）"
    elif estimation["峰值 QPS"] < 100000:
        estimation["缓存方案"] = "Redis Cluster（3-5 节点）"
    else:
        estimation["缓存方案"] = "多级缓存（本地缓存 + Redis Cluster）"
    
    return estimation

Level 4 · 边界与陷阱

4.1 系统设计面试中的常见算法错误

错误 1：选错数据结构的"层次"

# ❌ "用红黑树做数据库索引"
# 问题：红黑树是内存数据结构，每个节点一次 I/O
# 100 万行数据，树高 20 → 20 次随机磁盘 I/O → 200ms（假设 HDD）
# 
# ✅ "用 B+Tree 做数据库索引"
# B+Tree 阶=1000，100 万行数据，树高 2 → 2 次 I/O → 20ms

# ❌ "用 B+Tree 做内存数据库索引"
# 问题：B+Tree 为磁盘优化（大节点减少 I/O），内存中没有 I/O 开销
# 内存中 B+Tree 的 cache line 利用率反而不如红黑树/跳表
# 
# ✅ "用跳表做内存索引"（Redis 的选择）
# 或 "用 ART（Adaptive Radix Tree）做内存索引"（现代内存数据库的选择）

错误 2：混淆"精确"和"近似"的场景

# 场景：统计网站 UV（独立访客数）
# 
# ❌ "用 HashSet 存所有用户 ID"
# 问题：10 亿用户 × 8 bytes/ID = 8GB 内存
# 
# ✅ 分析：UV 需要精确值吗？
# - 如果用于计费（广告计费）→ 需要精确 → 用 bitmap + 分片
# - 如果用于数据分析仪表盘 → 接受 2% 误差 → HyperLogLog（12KB 内存！）

# Redis 中的 HyperLogLog
"""
PFADD page_views:2024-01-01 "user:12345"   # 添加
PFADD page_views:2024-01-01 "user:67890"   # 添加
PFCOUNT page_views:2024-01-01              # 获取 UV 估计值
# 标准误差 0.81%，内存固定 12KB，与基数大小无关！
"""

错误 3：忽略"热点问题"

# 场景：一致性哈希分布缓存
# 
# 问题：某个明星发了一条微博，数亿人访问 → 该 key 所在节点过载
# 一致性哈希无法解决热点（它只保证均匀分布，不保证均匀访问）
# 
# 解决方案：
HOT_KEY_SOLUTIONS = {
    "本地缓存": "热 key 识别后，在应用服务器本地缓存（TTL 短）",
    "Key 分散": "对热 key 加后缀（key_1, key_2, ...key_N），分散到 N 个节点",
    "读写分离": "热 key 专门用只读副本承接",
}

# 如何识别热 key？
# → Count-Min Sketch（概率数据结构，本书第 9 章）
# → 在网关层用 O(1) 空间估计每个 key 的访问频率

错误 4：在 AP 系统中追求强一致性

# 面试官："设计一个分布式购物车"
# 
# ❌ "用 Raft 保证购物车数据强一致"
# 问题：
# - 购物车不需要强一致（用户加了商品但暂时看不到，比服务不可用好）
# - Raft 在网络分区时少数派不可用 → 用户无法操作购物车
# 
# ✅ "用 CRDT（或 LWW）保证最终一致，AP 优先"
# - 网络分区时所有节点继续服务
# - 分区恢复后自动合并（CRDT 的冲突解决）
# - 极端情况下可能出现"删了又出现"（可接受，用户重新删除即可）

4.2 系统设计面试中如何"展示"算法知识

策略 1：自然引入，不生硬

# ❌ 生硬引入：
# "这里我要用一致性哈希。一致性哈希是 Karger 1997 年提出的..."
# （面试官：我知道什么是一致性哈希，你直接说怎么用）

# ✅ 自然引入：
# "数据分片用一致性哈希，每个物理节点 150 个虚拟节点。
#  这样节点增减时只需迁移约 1/N 的数据，避免缓存雪崩。
#  如果面试官追问，我可以解释虚拟节点为什么是 150 个。"

策略 2：用算法解释设计决策

# 面试官："为什么选 Redis 的 ZSET 而不是 MySQL 排序？"
# 
# 好的回答：
"""
两个原因，都和底层数据结构有关：

1. 写入性能：
   - ZSET 底层是跳表，更新分数 O(log n)，纯内存操作
   - MySQL 更新索引需要 B+Tree 的节点分裂，涉及磁盘 I/O
   - 在 QPS 10000+ 的排行榜场景，ZSET 延迟 < 1ms，MySQL 可能 > 10ms

2. 范围查询：
   - ZRANGEBYSCORE 沿跳表底层链表顺序扫描，O(log n + k)
   - MySQL 的 ORDER BY LIMIT 如果 offset 大，需要跳过前面的行
   - 深翻页时 ZSET 用 ZRANGEBYSCORE 的 cursor 模式更高效
"""

策略 3：主动提出 trade-off

# 面试官："如何实现全局唯一 ID？"
# 
# 展示多种方案及其算法背景：
UNIQUE_ID_TRADEOFFS = {
    "UUID v4": {
        "算法": "128-bit 随机数",
        "优点": "无需协调，完全去中心化",
        "缺点": "无序（B+Tree 索引频繁分裂）、128-bit 太长",
        "适用": "不需要有序性的场景",
    },
    "Snowflake": {
        "算法": "时间戳(41bit) + 机器ID(10bit) + 序列号(12bit) = 64bit",
        "优点": "趋势递增（B+Tree 友好）、64-bit 紧凑",
        "缺点": "依赖时钟（时钟回拨会重复）",
        "适用": "大多数互联网场景",
    },
    "数据库自增": {
        "算法": "中心化计数器",
        "优点": "严格递增、简单",
        "缺点": "单点瓶颈、扩展性差",
        "适用": "小规模系统、订单号",
    },
}

4.3 从算法面试到系统设计面试的思维转换

# 算法思维 → 系统设计思维的转换示例
# 
# 算法题："找到数组中出现次数最多的元素" → Counter + 排序
# 
# 系统设计版："实时统计全站最热门的 100 个搜索词"
# 
# 需要考虑的算法/数据结构：
# - Count-Min Sketch：O(1) 空间估计词频
# - Min-Heap (size=100)：维护 Top-100
# - 时间窗口：滑动窗口算法的分布式版本
# - 数据分区：按哈希分片到多台机器
# - 合并：Map-Reduce 思想，每台机器出 Top-100，再全局排序

4.4 真实系统案例：算法如何影响架构决策

案例 1：Google Bigtable 的布隆过滤器

"""
Bigtable (Chang et al., 2006) 的读路径：

1. 客户端请求读取某个 row key
2. 先查 MemTable（内存） — O(log n) 跳表查找
3. 如果不在 MemTable，需要查磁盘上的多个 SSTable
4. 每个 SSTable 有一个布隆过滤器：
   - BF 说"不在" → 跳过这个 SSTable（节省一次磁盘 I/O）
   - BF 说"可能在" → 读取该 SSTable
5. 实验数据：布隆过滤器将读操作的磁盘访问次数减少了 10x

算法选择的影响：
- 没有布隆过滤器：每次读需要查 N 个 SSTable → N 次随机 I/O
- 有布隆过滤器：期望只查 1 个 SSTable → 1 次随机 I/O
- 代价：每个 SSTable 额外存储 ~10 bits/key 的布隆过滤器
"""

案例 2：Kafka 的时间轮

"""
Kafka 需要管理数百万个延迟操作（超时请求、延迟 ACK 等）。

朴素方案：用优先队列（堆）
- 插入 O(log n)，删除最小 O(log n)
- n = 百万时，每次操作 20 次比较 → CPU 开销大

Kafka 的方案：层次时间轮（Hierarchical Timing Wheel）
- 插入 O(1)，到期检查 O(1)
- 通过溢出到高层轮处理长延迟

性能对比（Kafka 官方 benchmark）：
- 堆实现：100 万定时任务，吞吐约 50 万次/秒
- 时间轮：100 万定时任务，吞吐约 500 万次/秒（10x 提升）
"""

案例 3：Redis Cluster 的哈希槽 vs 一致性哈希

"""
Redis Cluster 没有用传统一致性哈希环，而是用了"哈希槽"（Hash Slot）：
- 固定 16384 个 slot
- CRC16(key) % 16384 → 得到 slot 编号
- 每个节点负责一部分 slot

为什么不用一致性哈希环？（antirez 的解释）
1. 哈希槽更容易管理：移动一个 slot = 移动一组 key
2. 配置信息更紧凑：16384 bit = 2KB 就能描述整个集群的 slot 分配
3. 一致性哈希的虚拟节点管理复杂

这是工程取舍的典型案例：
- 学术上一致性哈希更优雅
- 工程上哈希槽更简单、更可控
- 两者性能差异在 Redis 规模下可以忽略
"""

4.5 面试中的"系统设计 + 算法"融合题

近年来顶级公司出现了"融合题"——在系统设计面试中要求写算法代码：

# 融合题示例："设计一个限流器（Rate Limiter）"
# 
# 需要同时展示：
# 1. 系统设计：部署位置、分布式方案、容错
# 2. 算法实现：限流算法的精确实现

class SlidingWindowRateLimiter:
    """滑动窗口限流器
    
    系统设计考量：
    - 部署在 API Gateway 层
    - 用 Redis 存储窗口数据（多实例共享）
    - 时间窗口粒度选择：1秒 vs 1分钟 vs 自适应
    
    算法考量：
    - 固定窗口：简单但有边界突刺问题
    - 滑动日志：精确但内存大（存每个请求时间戳）
    - 滑动窗口计数器：折中方案（本实现）
    """
    
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = {}  # user_id → [(timestamp, count), ...]
    
    def allow_request(self, user_id: str, current_time: float) -> bool:
        """判断是否允许请求
        
        滑动窗口计数器算法：
        - 将时间划分为子窗口（如 1 秒一个）
        - 统计当前窗口内的请求总数
        - 用当前子窗口的位置做加权（处理窗口边界）
        """
        window_start = current_time - self.window_seconds
        
        if user_id not in self.requests:
            self.requests[user_id] = []
        
        # 清理过期记录
        self.requests[user_id] = [
            (ts, count) for ts, count in self.requests[user_id]
            if ts > window_start
        ]
        
        # 统计窗口内请求数
        total = sum(count for _, count in self.requests[user_id])
        
        if total >= self.max_requests:
            return False
        
        # 记录当前请求
        # 实际 Redis 实现：INCR + EXPIRE 的组合
        current_bucket = int(current_time)  # 按秒分桶
        for i, (ts, count) in enumerate(self.requests[user_id]):
            if int(ts) == current_bucket:
                self.requests[user_id][i] = (ts, count + 1)
                return True
        
        self.requests[user_id].append((current_time, 1))
        return True


class TokenBucket:
    """令牌桶算法 — 另一种限流实现
    
    与滑动窗口的对比：
    - 令牌桶允许突发流量（桶满时可以瞬间消耗）
    - 滑动窗口严格限制窗口内总量
    
    适用场景：
    - API 限流（允许短暂突发）→ 令牌桶
    - 防刷（严格限制频率）→ 滑动窗口
    """
    
    def __init__(self, rate: float, capacity: int):
        self.rate = rate        # 令牌填充速率（个/秒）
        self.capacity = capacity  # 桶容量
        self.tokens = capacity  # 当前令牌数
        self.last_refill_time = 0
    
    def allow_request(self, current_time: float) -> bool:
        """尝试消耗一个令牌"""
        # 补充令牌
        elapsed = current_time - self.last_refill_time
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill_time = current_time
        
        # 消耗令牌
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

本章小结

下一章预告： 第 43 章提供完整的学习路线和刷题指南，从入门到竞赛级别的进阶路径。