实战篇 / 第 26 章

第 26 章 · 面试题精选与学习路线

先看这一节：面试的底层逻辑与答题框架

技术面试本质上是信息不对称下的信任建立过程。面试官在有限的一小时里要回答一个问题：「这个候选人进来后，能不能独立把事情做对？」他无法验证你十年经验，只能靠几个精心设计的问题去探测你的深度、广度与边界感。

面试官真正在考什么

万能答题框架：STAR + FIVE

# 结构化表达模板（口述时在心里过一遍，不要背台词）

【1. 复述确认】 "我理解您想问的是 X，对吗？"
【2. 一句话结论】 "简单说，A 比 B 快，因为 X。"
【3. 分点展开】 "这里有三个关键点：
                     第一，……（机制/边界）
                     第二，……（实践中的坑）
                     第三，……（我实际怎么用的）"
【4. 反向验证】 "如果要我反过来选 B，条件是……"
【5. 主动补刀】 "还有个相关的问题您可能想知道：……"

# 遇到不会的问题
"这个我没在实际项目里用过。但按我的理解，它的核心是 X。
 如果让我实现，我会先……再…… 最后用…… 验证。
 我不确定的地方是 Y，您能提示一下方向吗？"

# 面试官问「你没做过的技术」
"这块我了解得比较浅。我目前的理解是 X。
 在上一个项目里，我用的是 Z，因为 Y。
 如果团队要用 X，我需要先补的是 ……"

面试前的知识查漏清单

1. 把你的简历上每一项技术都准备一个「我做了什么、遇到什么问题、怎么解决的」三句话版本。
2. 简历上写的框架，源码至少读过其入口、核心类、请求链路三处。
3. 准备 2 个你写砸过的项目复盘，这比 2 个成功案例更能体现能力。
4. 手边有可运行的 demo 或可讲的技术方案，比抽象描述有说服力得多。

语言基础高频题

这一组的共同特点：概念不难，但极易答得浅。能说出「是什么」只是及格，能说出「为什么这样设计、代价是什么」才是优秀。

1. 列表推导式、生成器表达式、集合推导式有什么区别？

结论：三者语法结构相同，区别在于结果的存储方式——列表推导生成一个真实 list，集合推导生成 set（去重），生成器表达式生成一个惰性求值的迭代器。

对比实验

import sys
import time

# 列表推导：立刻全部算完，全部存进内存
t0 = time.perf_counter()
squares_list = [x * x for x in range(1_000_000)]
print(f"列表推导 {time.perf_counter()-t0:.3f}s，占用 {sys.getsizeof(squares_list)/1024/1024:.1f} MB")
# 输出：列表推导 0.048s，占用 36.8 MB

# 生成器表达式：算一个用一个，不占内存
t0 = time.perf_counter()
squares_gen = (x * x for x in range(1_000_000))
print(f"生成器创建 {time.perf_counter()-t0:.6f}s，占用 {sys.getsizeof(squares_gen)} 字节")
# 输出：生成器创建 0.000002s，占用 48 字节

# 集合推导：去重 + 可快速查找
names = ["Bob", "bob", "BOB", "alice"]
print({n.lower() for n in names})        # {'bob', 'alice'}

# 关键区别：能否复用
gen = (i for i in range(3))
print(list(gen), list(gen))
# 输出：[0, 1, 2] []      —— 生成器是一次性的，第二次拿到空列表
print(list(squares_list)[:3], list(squares_list)[:3])
# 输出：[1, 4, 9] [1, 4, 9] —— 列表可以反复遍历

2. 可变默认参数为什么是经典陷阱？

结论：默认参数在函数定义时求值一次并绑定到函数对象上，之后所有调用共享同一个对象。

# 经典错误
def bad_append(item, bucket=[]):        # 坑在 [] 只在定义时创建一次
    bucket.append(item)
    return bucket

print(bad_append(1))    # [1]
print(bad_append(2))    # [1, 2]  ← 上次的结果还在！
print(bad_append(3))    # [1, 2, 3]

# 正确写法：用 None 做哨兵
def good_append(item, bucket=None):
    if bucket is None:
        bucket = []
    bucket.append(item)
    return bucket

print(good_append(1))   # [1]
print(good_append(2))   # [2]  ← 互不影响

# 更 Pythonic 的写法：用不可变默认 + *args 解包
def collect(item, *items, **kwargs):
    return [item, *items]

# 底层原理：默认值保存在函数对象的 __defaults__ 中
import inspect
print(inspect.signature(bad_append))
# 输出：(item, bucket=[])
print(bad_append.__defaults__)
# 输出：([],)     ← 全局共享的就是这一个列表对象

# 衍生考点：闭包中的循环变量
fns = [lambda: i for i in range(3)]
print([f() for f in fns])         # [2, 2, 2]  ← 都是最后一个 i
fns2 = [lambda i=i: i for i in range(3)]   # 默认参数绑定，逃逸技巧
print([f() for f in fns2])        # [0, 1, 2]

3. 深拷贝与浅拷贝的区别？

结论：浅拷贝只复制外层容器，内部对象仍是同一份引用；深拷贝递归复制所有层级，循环引用也能正确处理。

import copy

original = [[1, 2], [3, 4]]

# 赋值：连外层容器都是同一个对象
a = original
a[0].append(99)
print(original)            # [[1, 2, 99], [3, 4]]  ← 全被改了

# 浅拷贝：新外层，内层共享
original = [[1, 2], [3, 4]]
b = copy.copy(original)
b[0].append(99)
b.append([5])
print(original)            # [[1, 2, 99], [3, 4]]  ← 内层被改
print(b)                   # [[1, 2, 99], [3, 4], [5]]

# 深拷贝：全部复制
original = [[1, 2], [3, 4]]
c = copy.deepcopy(original)
c[0].append(99)
print(original)            # [[1, 2], [3, 4]]     ← 完全不受影响

# 深拷贝处理循环引用（这是它比手写递归强的地方）
lst = [1, 2]
lst.append(lst)                       # 自己引用自己
cloned = copy.deepcopy(lst)
print(cloned[2] is cloned)            # True —— 结构被正确重建

# copy 模块的完整规则表
class A:
    pass

print(copy.copy(1))                # 不可变类型直接返回原对象（1）
print(copy.copy("ab"))             # 'ab'  ← 不是列表的 copy 方法！
print(copy.copy((1, 2)))           # 元组不可变，返回自身
d = {"x": [1]}
print(copy.copy(d)["x"] is d["x"])     # True  浅：值共享
print(copy.deepcopy(d)["x"] is d["x"]) # False 深：值也复制

# 自定义类如何定义拷贝行为
import copyreg


class Resource:
    def __init__(self, handle):
        self.handle = handle          # 文件句柄等不可拷贝的资源

    def __copy__(self):
        return Resource(handle="shared-handle")

    def __deepcopy__(self, memo):
        return Resource(handle="independent-handle")


r = Resource("f1")
print(copy.deepcopy(r).handle)      # independent-handle

4. Python 的 GIL 是什么？它被取消了吗？

结论：GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器级别的互斥锁，保证同一时刻只有一个线程在执行字节码。因此多线程无法利用多核做 CPU 并行；但它不是语言的语义规定，只是 CPython 的实现细节，且从 Python 3.13 开始可以在部分平台编译成「自由线程（free-threaded）」版本，3.14 起逐步默认。

import multiprocessing as mp
import threading
import time


def cpu_work(n: int) -> int:
    total = 0
    for i in range(n):
        total += i * i
    return total


N = 8_000_000


def bench() -> None:
    t0 = time.perf_counter()
    threads = [threading.Thread(target=cpu_work, args=(N,)) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    t_thread = time.perf_counter() - t0

    t0 = time.perf_counter()
    with mp.Pool(4) as pool:
        pool.map(cpu_work, [N] * 4)
    t_proc = time.perf_counter() - t0

    print(f"4 线程(CPU 密集): {t_thread:.2f}s")
    print(f"4 进程(CPU 密集): {t_proc:.2f}s")
    print(f"进程/线程 加速比: {t_thread / t_proc:.1f}x")


if __name__ == "__main__":
    bench()
# 典型输出：
# 4 线程(CPU 密集): 2.60s
# 4 进程(CPU 密集): 0.72s
# 进程/线程 加速比: 3.6x  ← 线程几乎没有加速，因为 GIL 串行化了

# 反过来：IO 密集型任务，多线程非常有效
import urllib.request


def fetch(url: str) -> int:
    with urllib.request.urlopen(url, timeout=3) as r:
        return len(r.read())


# 20 个 URL 并发下载：多线程约 0.6s，串行约 12s
# 因为真正的耗时在网络等待，GIL 在等待期间是释放的

5. 装饰器的执行时机与本质

结论装饰器是「函数包装函数」。@log 修饰 def f(): 等价于 f = log(f)，在定义时执行一次，此后调用 f() 实际调用的是包装后的函数。

import functools
import time


# 1) 最简装饰器
def log_call(fn):
    def wrapper(*args, **kwargs):
        print(f"[调用] {fn.__name__} args={args}")
        result = fn(*args, **kwargs)
        print(f"[返回] {fn.__name__} -> {result}")
        return result
    return wrapper


# 2) 带参数与返回值的通用装饰器（用 *args 接收任意配置）
def retry(times: int = 3, delay: float = 0.1, exceptions=(Exception,)):
    def deco(fn):
        @functools.wraps(fn)                 # 保留原函数元信息
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as exc:
                    last_exc = exc
                    print(f"[重试] 第 {attempt}/{times} 次失败: {exc}")
                    if attempt < times:
                        time.sleep(delay * attempt)   # 指数退避
            raise last_exc
        return wrapper
    return deco


# 3) 组合使用：装饰器自下而上应用
calls = []


def count(fn):
    @functools.wraps(fn)
    def w(*a, **k):
        calls.append(fn.__name__)
        return fn(*a, **k)
    return w


@count
@retry(times=2)
def unstable(x: int) -> int:
    if x % 2 == 1:
        raise ValueError("随机失败")
    return x * 2


print(unstable(2))                 # 4
print(unstable(3))                 # 重试后抛 ValueError
print(calls)                       # ['unstable', 'unstable']

# 4) functools.wraps 为什么必须写
print(unstable.__name__)           # 'unstable'  ← 没有 wraps 就是 'wrapper'
print(unstable.__doc__)            # 原始 docstring 被保留

# 5) 类装饰器：可以给类加属性与方法
import functools


def add_repr(cls):
    cls.repr_short = lambda self: f"<{self.__class__.__name__} {self.id}>"
    return cls


@add_repr
class User:
    def __init__(self, uid):
        self.id = uid


print(User(7).repr_short())         # <User 7>

# 6) 参数装饰器的执行顺序验证
@print_msg("进入")
@print_msg("离开")
def demo():
    print("函数体")


# 定义阶段：demo 先被装饰一次
# 调用阶段：进入 -> 函数体 -> 离开（由内向外）

6. 迭代器与生成器

结论迭代器实现了 __iter__ 与 __next__，是「一次一步」的对象；生成器是「用 yield 快速写出的迭代器」，每次 yield 会把状态挂起，下次 next() 从挂起点恢复。

# 1) 自己实现一个迭代器协议
class Countdown:
    def __init__(self, start):
        self.current = start

    def __iter__(self):
        return self                  # 返回自身即为迭代器

    def __next__(self):
        if self.current <= 0:
            raise StopIteration       # 必须抛这个异常来结束
        self.current -= 1
        return self.current + 1

print(list(Countdown(3)))            # [3, 2, 1]

# 2) 生成器：yield 让函数变成「状态机」
def fib():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b


print([next(fib_gen) for _ in range(8)] if (fib_gen := fib()) else [])
# 输出：[0, 1, 1, 2, 3, 5, 8, 13]

# 3) send：双向通信，把值送进生成器内部
def echo():
    while True:
        received = yield              # yield 表达式先返回一个值，等下次 send
        print("收到:", received)


g = echo()
next(g)                            # 启动生成器，停在第一个 yield
g.send("你好")                     # 打印「收到: 你好」

# 4) 实战：分块读取大文件（生成器最典型的工程用途）
def read_in_chunks(path: str, size: int = 1024 * 64):
    with open(path, "r", encoding="utf-8") as f:
        while chunk := f.read(size):  # 海象运算符，3.8+
            yield chunk


# 5) 生成器实现内存友好的 CSV 统计
import csv
from collections import Counter


def count_words(path: str):
    """逐行流式统计，返回 Counter。10GB 文件也是常量内存。"""
    counter: Counter[str] = Counter()
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            counter.update(row["title"].lower().split())
    return counter


# 6) 陷阱：生成器只能消费一次
gen = (i for i in range(5))
print(list(gen))                    # [0, 1, 2, 3, 4]
print(list(gen))                    # []          ← 第二次为空
# 需要复用就转成 list 或 tuple，但会失去惰性优势

# 7) 双向链表 / 树的深度优先遍历常用生成器
def walk(node):
    yield node
    for child in node.children:
        yield from walk(child)       # 委托给子生成器

7. 面向对象：多继承、MRO 与魔术方法

结论Python 的 MRO（Method Resolution Order，方法解析顺序）遵循 C3 线性化算法，保证方法查找顺序既满足继承声明顺序，又在多继承下保持单调性（子类不会在父类之前）。

# 1) MRO 顺序
class A: pass
class B(A): pass
class C(A): pass
class D(B, C): pass

print(D.__mro__)
# 输出：(D, B, C, A, object)  —— B 在 C 前，因为 D 的声明顺序是 (B, C)

# 2) 菱形继承与 super() 的正确用法
class Logger:
    def __init__(self, name):
        self.name = name
        print(f"[Logger] 初始化 {name}")

    def report(self):
        print(f"[Logger] {self.name} 的日志")


class Database:
    def __init__(self, dsn):
        self.dsn = dsn
        print(f"[Database] 连接 {dsn}")


class Service(Database, Logger):       # 多继承混合能力
    def __init__(self, name, dsn):
        super().__init__(name)         # 只写一次，交给 MRO 决定谁先执行
        self.dsn = dsn

    def report(self):
        super().report()               # 链式调用所有父类实现
        print(f"[Service] 组合完成 dsn={self.dsn}")


Service("订单服务", "postgres://localhost/shop")
# 输出：
# [Logger] 初始化 订单服务      ← Logger 先跑（MRO 决定）
# [Service] 组合完成 dsn=postgres://localhost/shop

print(Service.__mro__)
# (Service, Database, Logger, object)

# 3) 抽象基类：强制子类实现
from abc import ABC, abstractmethod


class PaymentGateway(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        """所有子类必须实现"""

    @abstractmethod
    def refund(self, txn_id: str) -> bool: ...

    def describe(self) -> str:
        return f"支付网关: {type(self).__name__}"


# PaymentGateway()        # TypeError: Can't instantiate abstract class
class AlipayGateway(PaymentGateway):
    def pay(self, amount: float) -> str:
        return f"alipay_{amount}"

    def refund(self, txn_id: str) -> bool:
        return True


print(AlipayGateway().describe())     # 支付网关: AlipayGateway

# 4) 常用魔术方法对照
class Money:
    def __init__(self, amount: float):
        self.amount = amount

    def __repr__(self) -> str:      # 调试用，无歧义
        return f"Money({self.amount})"

    def __str__(self) -> str:       # 给人看
        return f"￥{self.amount:.2f}"

    def __eq__(self, other) -> bool:
        return isinstance(other, Money) and self.amount == other.amount

    def __hash__(self) -> int:      # 定义 __eq__ 后必须同时定义
        return hash(self.amount)      # 否则对象不可放入 set/dict

    def __lt__(self, other) -> bool:
        return self.amount < other.amount

    def __add__(self, other) -> "Money":
        return Money(self.amount + other.amount)

    def __bool__(self) -> bool:
        return self.amount > 0

    def __len__(self) -> int:
        return len(str(self.amount))

    def __getitem__(self, idx) -> str:
        return str(self)[idx]

    def __enter__(self) -> "Money":
        print("打开事务")
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        print("提交或回滚事务")
        return False                  # 返回 True 会吞掉异常


m = Money(19.9)
print(repr(m), str(m))                # Money(19.9) ￥19.90
print(m + Money(0.1))                 # Money(20.0)
print(bool(Money(0)))                 # False
print(m < Money(100))               # True
print(sorted([Money(3), Money(1)]))   # [Money(1), Money(3)]  依赖 __lt__


# 5) dataclass：现代 Python 写值对象的推荐方式
from dataclasses import dataclass, field, asdict
from typing import Optional


@dataclass(frozen=True, slots=True, order=True)   # 不可变 + 省内存 + 可比较
class User:
    uid: int
    name: str
    email: Optional[str] = None
    tags: list[str] = field(default_factory=list)


u = User(1, "张三")
print(asdict(u))                     # {'uid': 1, 'name': '张三', ...}
print(u == User(1, "张三"))           # True  自动生成 __eq__
print(User(1, "a") < User(2, "b")) # True  order=True 生成比较方法
try:
    u.uid = 99                       # frozen=True 抛异常
except Exception as e:
    print(type(e).__name__)          # FrozenInstanceError

数据结构与算法高频题

1. list、tuple、set、dict 的底层与复杂度

import sys
from collections import deque

# 1) list 的摊还扩容原理：容量按 1.125 倍增长
a = []
last_size = sys.getsizeof(a) - 56   # 减去对象头
for i in range(1000):
    a.append(i)
    cur = sys.getsizeof(a) - 56
    if cur != last_size:
        print(f"len={len(a):4d}  容量={cur//8:4d}")
        last_size = cur
# 输出（节选）：
# len=   5  容量=   8
# len=   9  容量=  16
# len=  17  容量=  32
# 观察：扩容不频繁，因此 append 均摊 O(1)；但单次扩容是 O(n) 的拷贝

# 2) 切片是 O(n) 且产生新对象，不是视图
xs = list(range(1_000_000))
t0 = __import__("time").perf_counter()
for _ in range(100):
    _ = xs[::2]          # 每次都完整复制
print(f"切片 100 次: {__import__('time').perf_counter()-t0:.3f}s")

# 3) deque 的正确用法：队列 / 栈 / 滑动窗口
q = deque(maxlen=5)       # maxlen 自动淘汰最旧的，空间有界
for i in range(10):
    q.append(i)
print(list(q))            # [5, 6, 7, 8, 9]

# 滑动窗口最大值（单调队列），O(n)
def max_in_window(nums: list[int], k: int) -> list[int]:
    dq = deque()                       # 存下标，保持值单调递减
    out = []
    for i, v in enumerate(nums):
        while dq and nums[dq[-1]] <= v:
            dq.pop()                  # 队尾不可能是最大值了
        dq.append(i)
        if dq[0] <= i - k:
            dq.popleft()              # 队首滑出窗口
        if i >= k - 1:
            out.append(nums[dq[0]])
    return out


print(max_in_window([1, 3, -1, -3, 5, 3, 6, 7], 3))   # [3, 3, 5, 5, 6, 7]

# 4) dict 的三个实用技巧
from collections import defaultdict, Counter, OrderedDict

dd = defaultdict(list)
for word in ["a", "b", "a"]:
    dd[word].append(1)                 # 不必先判断 key 是否存在
print(dict(dd))                       # {'a': [1, 1], 'b': [1]}

cnt = Counter("mississippi")           # 计数神器
print(cnt.most_common(2))              # [('i', 4), ('s', 4)]

od = OrderedDict([("b", 1), ("a", 2)])
od.move_to_end("b")                    # 普通 dict 做不到
print(list(od))                        # ['a', 'b']

# 5) 分组计数：dict + lambda 一步到位
data = [("北京", 18), ("上海", 15), ("北京", 20), ("上海", 12)]
grouped = {}
for city, age in data:
    grouped.setdefault(city, []).append(age)
print({k: sum(v) / len(v) for k, v in grouped.items()})
# {'北京': 19.0, '上海': 13.5}

2. 列表推导 vs 生成器 vs map/filter 的选型

# map/filter 是惰性的，返回迭代器
m = map(str, [1, 2, 3])
print(type(m))            # <class 'map'>
print(list(m))            # ['1', '2', '3']
print(list(m))            # []       一次性

# map 配合 zip 做并行处理
prices = [100.5, 20.0, 3.5]
rates = [0.9, 0.8, 1.0]
final = [round(p * r, 2) for p, r in zip(prices, rates, strict=True)]
print(final)              # [90.45, 16.0, 3.5]

# 生成器的提前退出优势：list 推导必须全部算完
def expensive(x):
    print(f"计算 {x}")
    return x * 2


squares = [expensive(x) for x in range(100)]   # 全部 100 个都会打印
print(len(squares))
# 输出：计算 0 ... 计算 99   然后才得到结果

g = (expensive(x) for x in range(100))         # 惰性
for i, v in enumerate(g):
    if i >= 3:
        break                  # 只算了 4 个就停
# 输出：计算 0..计算 3

# 三者性能实测
import timeit

nums = range(200_000)
t_list = timeit.timeit(lambda: [x * 2 for x in nums], number=10)
t_gen  = timeit.timeit(lambda: list(x * 2 for x in nums), number=10)
t_map  = timeit.timeit(lambda: list(map(lambda x: x * 2, nums)), number=10)
print(f"推导式 {t_list:.3f}s / 生成器 {t_gen:.3f}s / map {t_map:.3f}s")
# 典型：推导式最快，map 略慢（多一层 lambda 调用），差距不大

3. 常见排序与查找算法（手写一遍就够）

import random
from typing import Any, Callable


def bubble_sort(a: list[int]) -> list[int]:
    """O(n²)，教学意义 > 实用价值"""
    arr = a[:]
    for i in range(len(arr) - 1):
        swapped = False
        for j in range(len(arr) - 1 - i):
            if arr[j] > arr[j + 1]:
                arr[j], arr[j + 1] = arr[j + 1], arr[j]
                swapped = True
        if not swapped:            # 已有序，提前退出
            break
    return arr


def merge_sort(a: list[int]) -> list[int]:
    """O(n log n) 稳定排序：分治 + 归并，能用于「归并多个有序流」"""
    if len(a) <= 1:
        return a[:]
    mid = len(a) // 2
    left, right = merge_sort(a[:mid]), merge_sort(a[mid:])
    out, i, j = [], 0, 0
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:      # 用 <= 保证稳定性
            out.append(left[i]); i += 1
        else:
            out.append(right[j]); j += 1
    out.extend(left[i:]); out.extend(right[j:])
    return out


def quick_sort(a: list[int]) -> list[int]:
    """O(n log n) 平均 / O(n²) 最坏；三数取中可避免有序输入退化"""
    if len(a) <= 1:
        return a[:]
    pivot = a[len(a) // 2]
    left  = [x for x in a if x < pivot]
    mid   = [x for x in a if x == pivot]
    right = [x for x in a if x > pivot]
    return quick_sort(left) + mid + quick_sort(right)


def binary_search(a: list[int], target: int) -> int:
    """O(log n)，要求有序；找不到返回 -1"""
    lo, hi = 0, len(a) - 1
    while lo <= hi:
        mid = lo + (hi - lo) // 2       # 防溢出写法
        if a[mid] == target:
            return mid
        if a[mid] < target:
            lo = mid + 1
        else:
            hi = mid - 1
    return -1


def my_sort(items: list[int], key: Callable[[Any], Any] = None,
           reverse: bool = False) -> list[int]:
    """手写稳定排序 functools.cmp_to_key 版本"""
    from functools import cmp_to_key

    def cmp(a, b):
        ka, kb = (key(a), key(b)) if key else (a, b)
        return (ka < kb) - (ka > kb)

    return sorted(items, key=cmp_to_key(cmp), reverse=reverse)


def custom_sort(items: list[dict]) -> list[dict]:
    """多级排序：先按分数降序，同分按年龄升序"""
    return sorted(
        items,
        key=lambda d: (-d["score"], d["age"]),
    )


rows = [{"name": "a", "score": 90, "age": 20},
        {"name": "b", "score": 90, "age": 18},
        {"name": "c", "score": 95, "age": 25}]
print(custom_sort(rows))
# [{'name': 'c', ...}, {'name': 'b', ...}, {'name': 'a', ...}]

data = [5, 2, 9, 1, 7]
print(merge_sort(data))                  # [1, 2, 5, 7, 9]
print(binary_search(sorted(data), 7))    # 3

# 稳定性实测
pairs = [("a", 1), ("b", 2), ("c", 3)]
print(sorted(pairs, key=lambda p: p[0]))          # 稳定
print(sorted(pairs, key=lambda p: -p[0]))         # 反转键仍稳定：b, c, a

4. LRU 缓存与 collections

from collections import OrderedDict, deque, Counter, defaultdict, namedtuple
from functools import lru_cache, cache

# 1) 手写 LRU：用 OrderedDict 的「移动到末尾」能力 O(1)
class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()

    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)        # 标记为最近使用
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # 淘汰最久未用的

    def __len__(self):
        return len(self.cache)


c = LRUCache(2)
c.put(1, "a"); c.put(2, "b")
print(c.get(1))       # a（1 变为最近使用）
c.put(3, "c")          # 容量满，淘汰 2
print(c.get(2))       # -1

# 2) 内置 functools.lru_cache（线程安全的 C 实现）
calls = []


@lru_cache(maxsize=128)
def fetch_user(uid: int) -> dict:
    calls.append(uid)
    return {"id": uid}


fetch_user(1); fetch_user(1)
print(f"真实查询 {len(calls)} 次")               # 1
print(fetch_user.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
fetch_user.cache_clear()
fetch_user.cache_parameters()                 # {'maxsize': 128, 'typed': False}

# 3) cache：无上限版本（3.9+）
@cache
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)

print(fib(60))                                 # 秒出
fib.cache_clear()

# 4) 带缓存的方法要小心：self 是参数，导致整个对象被强引用
class Service:
    def __init__(self, dsn):
        self.dsn = dsn

    @lru_cache
    def heavy_query(self, key):               # self 在缓存键里
        return key * 2

# 更好：缓存挂在实例属性上，或确保服务对象生命周期明确

# 5) namedtuple 与 dataclass 的选择
Point = namedtuple("Point", "x y")            # 不可变、可索引、体积小
p = Point(1, 2)
print(p.x, p[0], tuple(p))                    # 1 1 (1, 2)

# 6) deque 的堆用法：优先队列
import heapq

heap = []
for x in [5, 1, 8]:
    heapq.heappush(heap, x)
print(heapq.heappop(heap))                    # 1  最小堆，O(log n)

# heapq 没有 maxheap，用负数模拟
maxh = [-x for x in [5, 1, 8]]
heapq.heapify(maxh)
print(-heapq.heappop(maxh))                   # 8

并发编程高频题

1. 多线程 vs 多进程 vs 协程，选型标准是什么？

import asyncio
import time

# 1) asyncio 版：100 个任务并发，单线程搞定
async def fetch(url_id: int) -> int:
    await asyncio.sleep(0.1)                 # 模拟网络等待
    return url_id


async def main_async():
    t0 = time.perf_counter()
    results = await asyncio.gather(*[fetch(i) for i in range(100)])
    dt = time.perf_counter() - t0
    print(f"asyncio 100 个任务: {dt:.2f}s，结果和={sum(results)}")
    # 输出：asyncio 100 个任务: 0.10s，结果和=4950


asyncio.run(main_async())                    # 串行会是 10s

# 2) 控制并发量：别一次创建 10 万个任务，会打爆内存与 fd
async def limited_main():
    sem = asyncio.Semaphore(20)              # 信号量限流

    async def guarded(i: int) -> int:
        async with sem:                      # 最多 20 个同时执行
            await asyncio.sleep(0.05)
            return i

    return await asyncio.gather(*[guarded(i) for i in range(200)])


print(len(asyncio.run(limited_main())))      # 200

# 3) 同步库放进 async 的三种处理方式
import httpx
import asyncio


async def with_native_async():
    """最佳：换用原生异步库"""
    async with httpx.AsyncClient() as client:
        r = await client.get("https://example.com")
        return r.status_code


async def with_threadpool():
    """次优：把阻塞调用丢进线程池，不阻塞事件循环"""
    loop = asyncio.get_running_loop()            # 必须在协程内获取

    def blocking(url):                            # 普通同步函数
        return httpx.get(url, timeout=10)         # 同步客户端调用

    resp = await loop.run_in_executor(None, blocking, "https://example.com")
    return resp.status_code


# 4) asyncio 的三个常用原语对比
async def primitives_demo():
    # TaskGroup（3.11+）：一组任务全成功才结束，一个失败自动取消其余
    async def work(i: int) -> int:
        await asyncio.sleep(0.01)
        return i * 2

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(work(i)) for i in range(5)]
    print("TaskGroup 结果:", [t.result() for t in tasks])

    # 超时控制
    try:
        await asyncio.wait_for(asyncio.sleep(5), timeout=0.1)
    except asyncio.TimeoutError:
        print("wait_for 超时了")

    # 取消与清理
    async def cancellable():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("收到取消信号，清理资源")
            raise

    task = asyncio.create_task(cancellable())
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass


asyncio.run(primitives_demo())

2. 线程安全：锁、条件变量、队列

import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 1) 最经典：非原子操作导致的竞态条件
counter = 0
lock = threading.Lock()


def unsafe_increment():
    global counter
    for _ in range(100_000):
        counter += 1          # 读-改-写三步，不是原子的


def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1


def run_bench(fn):
    global counter
    counter = 0
    threads = [threading.Thread(target=fn) for _ in range(4)]
    t0 = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return counter, time.perf_counter() - t0


print("无锁:", run_bench(unsafe_increment))   # 计数器明显少于 400000
print("有锁:", run_bench(safe_increment))

# 2) RLock：可重入锁，避免递归调用自身时死锁
rlock = threading.RLock()


def recurse(n: int) -> int:
    with rlock:
        return 1 if n == 0 else recurse(n - 1) + 1


print(recurse(100))                            # 没有 RLock 会死锁

# 3) Condition：生产者-消费者
import queue

data_queue: queue.Queue[int] = queue.Queue(maxsize=10)
done = threading.Event()


def consumer():
    while True:
        item = data_queue.get()
        if item is None:                        # 哨兵值表示结束
            data_queue.task_done()
            return
        print(f"消费 {item}")
        data_queue.task_done()


def producer(n: int):
    for i in range(n):
        data_queue.put(i)
    data_queue.put(None)


consumers = [threading.Thread(target=consumer) for _ in range(2)]
for c in consumers:
    c.start()
producer(10)
data_queue.join()                               # 等待所有 task_done
done.set()
for c in consumers:
    c.join()

# 4) 优先用 Queue 而不是自己拿锁
def worker(q: queue.Queue, results: list):
    while True:
        job = q.get()
        if job is None:
            q.task_done()
            break
        results.append(job * job)
        q.task_done()


jobs: queue.Queue = queue.Queue()
results: list[int] = []
for i in range(10):
    jobs.put(i)
for _ in range(3):
    jobs.put(None)
threads = [threading.Thread(target=worker, args=(jobs, results)) for _ in range(3)]
for t in threads:
    t.start()
jobs.join()
print(sorted(results))                          # 线程安全的收集

# 5) ThreadPoolExecutor：生产环境最常用的线程池
def heavy_io(task_id: int) -> str:
    time.sleep(0.1)
    return f"任务{task_id}完成"


with ThreadPoolExecutor(max_workers=10, thread_name_prefix="io") as pool:
    futures = {pool.submit(heavy_io, i): i for i in range(20)}
    done_count = 0
    for fut in concurrent.futures.as_completed(futures):
        if fut.exception():
            print(f"任务 {futures[fut]} 失败: {fut.exception()}")
        else:
            done_count += 1
print(f"完成 {done_count}/20")

# 6) GIL 下的「假并行」优化：把纯计算交给 C 层
from functools import lru_cache


def fast_way(xs: list[int]) -> list[int]:
    return list(map(abs, xs))                   # map 内部是 C 循环，持锁时间极短


def slow_way(xs: list[int]) -> list[int]:
    return [abs(x) for x in xs]                 # 纯 Python 循环，频繁争锁


big = list(range(100_000)) * 10
print(timeit.timeit(lambda: fast_way(big), number=5))
print(timeit.timeit(lambda: slow_way(big), number=5))

3. multiprocessing 的进程池与通信

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, TimeoutError as FTimeout
import os
import time


def compute(n: int) -> int:
    """子进程执行的函数：必须可 pickle（顶层函数，不能是 lambda 或局部函数）"""
    return sum(i * i for i in range(n))


def init_worker() -> None:
    """每个子进程启动时执行一次：建连接池在这里做最划算"""
    import redis
    global _conn
    _conn = redis.from_url("redis://localhost:6379/0")


def task_with_init(x: int) -> int:
    return x + 1


if __name__ == "__main__":
    # 1) 基本用法：map 分发，分块计算
    with mp.Pool(processes=4) as pool:
        t0 = time.perf_counter()
        results = pool.map(compute, [1_000_000] * 8)
        print(f"进程池 8 个任务: {time.perf_counter()-t0:.2f}s")

    # 2) imap：惰性返回，适合流式处理与提前终止
    with mp.Pool(4) as pool:
        for i, r in enumerate(pool.imap(compute, [100_000] * 100, chunksize=5)):
            if i >= 5:
                pool.terminate()              # 不要了就主动终止
                break
            print(i, r)

    # 3) 起新式接口 ProcessPoolExecutor：更好的返回值与异常传递
    with ProcessPoolExecutor(max_workers=4, initializer=init_worker) as ex:
        fut = ex.submit(task_with_init, 1)
        print(fut.result(timeout=5))
        print(list(ex.map(task_with_init, range(5))))

    # 4) 进程间通信四种方式
    # (a) Queue：生产者-消费者
    q = mp.Queue()
    p = mp.Process(target=lambda: q.put("来自子进程"))
    p.start()
    print(q.get())
    p.join()

    # (b) Pipe：点对点双向
    parent, child = mp.Pipe()
    p = mp.Process(target=lambda: child.send("ping"))
    p.start()
    print(parent.recv())
    p.join()

    # (c) SharedMemory：大数组高效共享（比 Manager 快得多）
    from multiprocessing import shared_memory
    shm = shared_memory.SharedMemory(create=True, size=1024 * 64)
    buf = shm.buf
    buf[:5] = b"hello"
    print(bytes(buf[:5]))                     # b'hello'
    shm.close(); shm.unlink()

    # (d) Manager：代理对象，适合复杂结构但慢
    with mp.Manager() as mgr:
        d = mgr.dict({"a": 1})
        d["b"] = 2
        print(dict(d))

    # 5) 平台差异：Windows/macOS 的 spawn 必须 if __name__ 保护
    print(mp.get_start_method())               # Linux: fork；Windows: spawn

    # 6) fork 带来的坑：子进程继承了父进程的锁与连接
    # 典型症状：fork 后数据库连接在两个进程共用同一 socket，导致协议错乱
    # 解决：用 spawn 上下文，或在 fork 前建立连接、fork 后重建
    ctx = mp.get_context("spawn")
    with ctx.Pool(2) as pool:
        print(pool.map(compute, [1000, 2000]))

4. 死锁的成因与规避

# 死锁四个必要条件（Coffman 条件）
# 1) 互斥   2) 持有并等待   3) 不可剥夺   4) 循环等待

# 经典死锁：两个线程以相反顺序获取两把锁
import threading

lock_a = threading.Lock()
lock_b = threading.Lock()


def transfer_a():
    with lock_a:                  # 先拿 A
        with lock_b:              # 等 B
            pass


def transfer_b():
    with lock_b:                  # 先拿 B —— 顺序相反
        with lock_a:              # 等 A -> 死锁
            pass


# 解法一：全局固定加锁顺序
import itertools
ORDER = (lock_a, lock_b)


def transfer_fixed(which_from, which_to, amount):
    with which_from:
        with which_to:
            pass


# 解法二：一次性申请所有锁（用 acquire 的多参数形式）
ok = lock_a.acquire(blocking=False) and lock_b.acquire(blocking=False)
if ok:
    try:
        pass
    finally:
        lock_b.release(); lock_a.release()

# 解法三：加超时
if lock_a.acquire(timeout=1.0):
    try:
        if lock_b.acquire(timeout=1.0):
            try:
                pass
            finally:
                lock_b.release()
    finally:
        lock_a.release()

# 解法四：用 contextlib 做可重入与有序锁
import contextlib
from contextlib import contextmanager


@contextmanager
def ordered_locks(*locks):
    """按 id 排序获取，彻底规避循环等待"""
    for lk in sorted(locks, key=lambda x: id(x)):
        lk.acquire()
    try:
        yield
    finally:
        for lk in reversed(sorted(locks, key=lambda x: id(x))):
            lk.release()


# 解法五：尽量用 Queue 传递数据，而不是锁保护共享状态
# 生产环境最实用的一条：能用消息队列解决的并发，不要用锁

数据库与存储高频题

1. ACID 与隔离级别

import sqlite3
import threading

# SQLite 支持串行化隔离，用它演示并发读写的差异
conn = sqlite3.connect(":memory:", check_same_thread=False)
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("CREATE TABLE account (id INTEGER PRIMARY KEY, balance INTEGER)")
conn.execute("INSERT INTO account VALUES (1, 1000)")

# 1) 原子更新：一条 SQL 完成「读改写」，避免丢失更新
def safe_deduct(uid: int, amount: int) -> bool:
    cur = conn.cursor()
    cur.execute(
        "UPDATE account SET balance = balance - ? "
        "WHERE id = ? AND balance >= ?",     # 条件写进 SQL，靠数据库保证原子
        (amount, uid, amount),
    )
    conn.commit()
    return cur.rowcount > 0                  # 影响行数为 0 说明余额不足


print(safe_deduct(1, 300))                     # True
print(conn.execute("SELECT balance FROM account WHERE id=1").fetchone())  # (700,)

# 2) 不安全的写法：先查后改，事务间存在竞态
def unsafe_deduct(uid: int, amount: int) -> bool:
    row = conn.execute("SELECT balance FROM account WHERE id=?", (uid,)).fetchone()
    if row[0] < amount:
        return False
    time.sleep(0.001)                          # 模拟两次 SQL 之间的耗时
    conn.execute("UPDATE account SET balance=? WHERE id=?", (row[0] - amount, uid))
    conn.commit()
    return True


# 3) 悲观锁 vs 乐观锁
# 悲观锁：SELECT ... FOR UPDATE，串行化冲突，适合冲突频繁
def pessimistic_deduct(uid: int, amount: int) -> bool:
    conn.execute("BEGIN")
    row = conn.execute(
        "SELECT balance FROM account WHERE id=? FOR UPDATE", (uid,)
    ).fetchone()
    if row[0] < amount:
        conn.rollback()
        return False
    conn.execute("UPDATE account SET balance=? WHERE id=?", (row[0] - amount, uid))
    conn.commit()
    return True


# 4) 乐观锁：版本号，适合冲突少
# 建表时：CREATE TABLE account (..., version INTEGER DEFAULT 0);
def optimistic_deduct(uid: int, amount: int, expected_version: int) -> bool:
    cur = conn.execute(
        "UPDATE account SET balance = balance - ?, version = version + 1 "
        "WHERE id = ? AND version = ?",         # 版本不匹配则影响 0 行
        (amount, uid, expected_version),
    )
    conn.commit()
    return cur.rowcount > 0


# 5) 显式事务：Python DB-API 的事务边界
conn.execute("BEGIN")
try:
    conn.execute("UPDATE account SET balance = balance + 100 WHERE id = 1")
    conn.execute("UPDATE account SET balance = balance - 100 WHERE id = 2")
    conn.commit()                               # 全部成功才提交
except Exception:
    conn.rollback()                             # 任一失败整体回滚
    raise

# 6) 连接池的正确使用：不要每次请求新建连接
import contextlib


@contextlib.contextmanager
def get_conn(pool):
    """从池里借连接，用完必还；即使出错也要归还"""
    conn = pool.getconn()
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    finally:
        pool.putconn(conn)

2. 索引原理：B+ 树、聚簇索引与最左前缀

# 1) 最左前缀与范围截断
-- 建联合索引
CREATE INDEX idx_abc ON orders (user_id, status, created_at);

-- ✅ 走索引
SELECT * FROM orders WHERE user_id = 1;
SELECT * FROM orders WHERE user_id = 1 AND status = 'paid';
SELECT * FROM orders WHERE user_id = 1 AND status = 'paid' AND created_at > '2026-01-01';

-- ⚠️ 部分可用：user_id 走了索引，created_at 没用上
SELECT * FROM orders WHERE user_id = 1 AND created_at > '2026-01-01';

-- ❌ 不走索引：跳过了最左列
SELECT * FROM orders WHERE status = 'paid';

-- ❌ 不走索引：前导通配符
SELECT * FROM orders WHERE user_id LIKE '%123';

# 2) 覆盖索引：避免回表
-- 需要字段都在索引里，EXPLAIN 的 Extra 出现 Using index
CREATE INDEX idx_cover ON orders (user_id, status, amount);
SELECT user_id, status, amount FROM orders WHERE user_id = 1;
-- 只查索引里的列，不需要回聚簇索引 -> 快得多

# 3) 排序方向与索引方向一致时可用索引消除排序
CREATE INDEX idx_time ON orders (created_at DESC);
SELECT * FROM orders ORDER BY created_at DESC LIMIT 20;

# 4) 索引下推（Index Condition Pushdown，ICP）
SELECT * FROM orders WHERE user_id = 1 AND name LIKE '%张%';
-- ICP 把 name 的过滤下推到存储引擎层，减少回表次数

3. 深分页、慢查询与连接池

# 1) 深分页问题
-- ❌ OFFSET 1000000 LIMIT 20：数据库要先扫描并丢弃前 100 万行
SELECT * FROM orders ORDER BY id LIMIT 1000000, 20;

-- ✅ 方案一：游标分页（keyset pagination），只扫 20 行
SELECT * FROM orders WHERE id > 12345 ORDER BY id LIMIT 20;

-- ✅ 方案二：业务排序字段 + 游标
SELECT * FROM orders
WHERE (created_at, id) < ('2026-01-01 10:00:00', 999)
ORDER BY created_at DESC, id DESC
LIMIT 20;

# Python 侧实现游标分页
import sqlite3

conn = sqlite3.connect("shop.db")
PAGE_SIZE = 20


def fetch_page(cursor: tuple[str, int] | None = None) -> tuple[list, tuple | None]:
    if cursor is None:
        sql = "SELECT created_at, id, amount FROM orders ORDER BY created_at DESC, id DESC LIMIT ?"
        args = (PAGE_SIZE,)
    else:
        last_time, last_id = cursor
        sql = ("SELECT created_at, id, amount FROM orders "
               "WHERE (created_at, id) < (?, ?) "
               "ORDER BY created_at DESC, id DESC LIMIT ?")
        args = (last_time, last_id, PAGE_SIZE)

    rows = conn.execute(sql, args).fetchall()
    if not rows:
        return [], None
    last = rows[-1]
    return rows, (last[0], last[1])


rows, next_cursor = fetch_page()
while rows and next_cursor:
    rows, next_cursor = fetch_page(next_cursor)

# 2) 慢查询日志
# MySQL: SET GLOBAL long_query_time = 0.2;
#        SHOW VARIABLES LIKE 'slow_query_log%';
# PostgreSQL: ALTER SYSTEM SET log_min_duration_statement = 200;
#            SELECT * FROM pg_stat_statements ORDER BY total_exec_time DESC LIMIT 10;

# 3) 用 Python 做慢查询定位与告警
import logging
import time

slow_queries: list[tuple[float, str]] = []
SLOW_MS = 200


def timed_query(sql: str, params: tuple = (), conn=conn) -> list:
    t0 = time.perf_counter()
    cur = conn.execute(sql, params)
    rows = cur.fetchall()
    dt = (time.perf_counter() - t0) * 1000
    if dt > SLOW_MS:
        slow_queries.append((dt, sql))
        logging.warning("慢查询 %.1fms: %s params=%s", dt, sql, params)
        if len(slow_queries) > 100:
            logging.error("检测到 %d 条慢查询，检查索引与执行计划", len(slow_queries))
    return rows


timed_query("SELECT * FROM orders WHERE user_id = ?", (1,))

# 4) 连接池：数据库能承受的连接数计算
# 连接数上限 = CPU 核数 * 2 + 有效磁盘数（PostgreSQL 官方建议）
# 应用侧：每个 gunicorn worker 独立建池
#   总连接数 = workers 数 × pool_size
# 例：8 workers × pool_size 10 + 20 个管理连接 = 100 -> 需要 max_connections >= 120
import os

WORKERS = 8
POOL_SIZE = 10
print(f"总连接数 = {WORKERS} × {POOL_SIZE} = {WORKERS * POOL_SIZE}")
print(f"建议 max_connections = {WORKERS * POOL_SIZE + 20}")

# 5) 连接池参数的工程含义
# pool_size      : 常驻连接数，应 >= (该进程并发请求数 / worker 数)
# max_overflow   : 峰值临时连接，超过后请求排队并超时
# pool_timeout   : 排队等待秒数，超时抛 QueuePool timeout —— 这个异常几乎总是
#                  「数据库慢」的信号，而不是「应用配置错」

4. Redis 高频数据结构与缓存一致性

import time
import redis

r = redis.from_url("redis://localhost:6379/0", decode_responses=True)

# 1) String 做分布式锁（SET NX EX）
LOCK = "lock:order:1001"


def acquire_lock(key: str, ttl_ms: int = 5000) -> str | None:
    token = __import__("secrets").token_urlsafe(16)      # 唯一标识，防误删他人锁
    ok = r.set(key, token, nx=True, ex=ttl_ms // 1000)
    return token if ok else None


def release_lock(key: str, token: str) -> None:
    """必须用 Lua 保证「比对并删除」的原子性"""
    lua = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    r.eval(lua, 1, key, token)


# 2) ZSet 做排行榜
def record_score(user: str, score: float) -> None:
    r.zadd("rank:score", {user: score})


def top_n(n: int = 10) -> list[tuple[str, float]]:
    return r.zrevrange("rank:score", 0, n - 1, withscores=True)


def user_rank(user: str) -> int | None:
    rank = r.zrevrank("rank:score", user)          # 0-based
    return rank + 1 if rank is not None else None


record_score("alice", 900)
record_score("bob", 750)
record_score("carol", 1200)
print(top_n(3))          # [('carol', 1200.0), ('alice', 900.0), ('bob', 750.0)]

# 3) ZSet 做延时队列（比 Redis List 靠谱）
def delay_task(task_id: str, delay_seconds: int) -> None:
    now = time.time()
    r.zadd("delay:queue", {task_id: now + delay_seconds})


def poll_due_tasks(limit: int = 100) -> list[str]:
    now = time.time()
    due = r.zrangebyscore("delay:queue", 0, now, start=0, num=limit)
    if due:
        r.zrem("delay:queue", *due)               # 取出的同时删除
    return due


delay_task("order_timeout:1001", 1800)

# 4) HyperLogLog 统计 UV：100 万用户只用 ~12KB
pf = r.pfadd("uv:20260101", *[f"user{i}" for i in range(1_000_000)])
print(pf)                                          # 1 表示已创建
print(r.pfcount("uv:20260101"))                   # 约 997307（误差 < 1%）

# 5) Pipeline 批处理：把 100 次往返压成 1 次
import time as _t

t0 = _t.perf_counter()
for i in range(1000):
    r.set(f"k{i}", i)
print(f"逐条 1000 次: {_t.perf_counter()-t0:.3f}s")

t0 = _t.perf_counter()
pipe = r.pipeline(transaction=False)             # transaction=False 更快
for i in range(1000):
    pipe.set(f"k{i}", i)
pipe.execute()
print(f"Pipeline 1000 次: {_t.perf_counter()-t0:.3f}s")   # 通常快 5~10 倍

# 6) Lua 保证复杂操作的原子性
LUA = """
local stock = tonumber(redis.call('get', KEYS[1]) or '0')
local need = tonumber(ARGV[1])
if stock >= need then
    redis.call('decrby', KEYS[1], need)
    return 1
else
    return 0
end
"""
if r.eval(LUA, 1, "stock:sku1", 1):
    print("扣减成功")
else
    print("库存不足")

Web 与工程化高频题

1. WSGI、ASGI 与框架关系

# WSGI 应用的最简形态：一个可调用对象，接收 environ 与 start_response
def simple_wsgi_app(environ, start_response):
    """environ 包含请求的全部信息（PATH_INFO、QUERY_STRING、REQUEST_METHOD 等）"""
    path = environ.get("PATH_INFO", "/")
    method = environ.get("REQUEST_METHOD", "GET")

    if path == "/healthz" and method == "GET":
        body = b'{"status":"ok"}'
        status = "200 OK"
    elif path == "/echo" and method == "POST":
        try:
            length = int(environ.get("CONTENT_LENGTH") or 0)
        except ValueError:
            length = 0
        raw = environ["wsgi.input"].read(length)
        body = raw.upper() if raw else b""
        status = "200 OK"
    else:
        body, status = b"not found", "404 Not Found"

    headers = [("Content-Type", "application/octet-stream"),
               ("Content-Length", str(len(body)))]
    start_response(status, headers)              # 必须先调用 start_response
    return [body]                                 # 必须返回可迭代的 bytes 序列


# 用 wsgiref 起一个测试服务器
from wsgiref.simple_server import make_server

if __name__ == "__main__":
    srv = make_server("127.0.0.1", 8001, simple_wsgi_app)
    print("测试服务器启动：http://127.0.0.1:8001/healthz")
    # srv.serve_forever()

# ASGI 应用的最简形态：async 可调用对象，接收 scope/receive/send
async def simple_asgi_app(scope, receive, send):
    """scope 是连接信息，receive/send 是协程，用来收发消息"""
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})
                return
        return

    if scope["type"] != "http":
        return

    await send({"type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"text/plain; charset=utf-8")]})
    await send({"type": "http.response.body", "body": b"hello from asgi"})


# 三层结构总结：
#   客户端 -> Nginx -> WSGI/ASGI 服务器(gunicorn/uvicorn) -> 应用框架 -> 你的代码

2. 常见中间件要做什么

import time
import uuid
from functools import wraps

from flask import Flask, g, request, jsonify

app = Flask(__name__)


# 1) 请求 ID 中间件：贯穿日志与下游调用
@app.before_request
def _assign_request_id():
    g.request_id = request.headers.get("X-Request-Id", uuid.uuid4().hex)
    g.start = time.perf_counter()


@app.after_request
def _add_headers(resp):
    resp.headers["X-Request-Id"] = g.request_id
    resp.headers["X-Response-Time-Ms"] = f"{(time.perf_counter()-g.start)*1000:.2f}"
    return resp


# 2) 统一错误处理中间件（装饰器版，可挂到任意接口）
def handle_errors(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except ValidationError as exc:
            return jsonify(code=400, message="参数错误", detail=exc.errors), 400
        except NotFoundError as exc:
            return jsonify(code=404, message=str(exc)), 404
        except PermissionError as exc:
            return jsonify(code=403, message=str(exc)), 403
        except Exception:
            app.logger.exception("未处理异常 request_id=%s", g.get("request_id"))
            # 注意：生产环境绝不返回 traceback
            return jsonify(code=500, message="服务内部错误"), 500
    return wrapper


class ValidationError(Exception): pass
class NotFoundError(Exception): pass


@app.get("/api/orders/<oid>")
@handle_errors
def get_order(oid: int):
    if oid != 42:
        raise NotFoundError(f"订单 {oid} 不存在")
    return jsonify(id=oid, status="paid")


# 3) 限流中间件：滑动窗口或令牌桶
class RateLimiter:
    """内存版滑动窗口：适合单实例；多实例请用 Redis 实现"""

    def __init__(self, limit: int = 100, window: int = 60):
        self.limit, self.window = limit, window
        self.hits: dict[str, list[float]] = {}

    def check(self, key: str) -> bool:
        now = time.time()
        window_hits = [t for t in self.hits.get(key, []) if now - t < self.window]
        if len(window_hits) >= self.limit:
            return False
        window_hits.append(now)
        self.hits[key] = window_hits
        return True

    def cleanup(self) -> None:
        now = time.time()
        for key in list(self.hits):
            self.hits[key] = [t for t in self.hits[key] if now - t < self.window]
            if not self.hits[key]:
                del self.hits[key]


limiter = RateLimiter(limit=10, window=60)


@app.before_request
def _rate_limit():
    if request.endpoint == "health":
        return None
    ip = request.remote_addr or "unknown"
    if not limiter.check(ip):
        return jsonify(code=429, message="请求过于频繁"), 429
    return None


@app.get("/health")
def health():
    return jsonify(status="ok")


# 4) 幂等中间件：防止重复提交
_seen: dict[str, object] = {}


def idempotent(key_prefix: str = "idemp"):
    def deco(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            key = f"{key_prefix}:{request.headers.get('Idempotency-Key', '')}"
            if key in _seen:
                return jsonify(code=0, data=_seen[key], duplicated=True), 200
            result = fn(*args, **kwargs)
            _seen[key] = result.get_json() if hasattr(result, "get_json") else result
            return result
        return wrapper
    return deco


@app.post("/api/orders")
@idempotent()
def create_order():
    return jsonify(order_no="ORD20260101001", amount=99.0)

3. RESTful、状态码与 API 设计

from dataclasses import dataclass
from datetime import datetime
from enum import Enum


class OrderStatus(str, Enum):
    """用 str 混入 Enum：序列化时自动转字符串，且可比较"""
    CREATED = "created"
    PAID = "paid"
    SHIPPED = "shipped"
    COMPLETED = "completed"
    CANCELLED = "cancelled"

    @property
    def is_final(self) -> bool:
        return self in {OrderStatus.COMPLETED, OrderStatus.CANCELLED}


@dataclass
class ApiResponse:
    code: int
    message: str
    data: object = None
    request_id: str | None = None

    def to_dict(self) -> dict:
        d = {"code": self.code, "message": self.message, "data": self.data}
        if self.request_id:
            d["request_id"] = self.request_id
        return d


# 分页与排序参数的安全处理
import re

SORT_WHITELIST = {"created_at", "amount", "status"}


def parse_pagination(args: dict) -> tuple[int, int, str, str]:
    page = max(1, min(int(args.get("page", 1)), 1000))          # 上限防爬
    size = max(1, min(int(args.get("size", 20)), 100))          # 上限防压库
    sort = args.get("sort", "created_at")
    if sort not in SORT_WHITELIST:                             # 白名单防注入
        sort = "created_at"
    order = "desc" if args.get("order", "desc").lower() == "desc" else "asc"
    return page, size, sort, order


# 分页响应的标准结构
def build_page(rows: list, total: int, page: int, size: int) -> dict:
    return ApiResponse(
        code=0, message="ok", data={
            "items": rows,
            "pagination": {
                "page": page, "size": size, "total": total,
                "pages": (total + size - 1) // size,
            },
        },
    ).to_dict()


# 接口版本化策略
# /api/v1/orders   —— 显式版本，改动大时升 v2
# /orders          —— 通过 Accept 头或请求头协商
# 实践：内部系统用头部协商，开放 API 用路径版本

4. 测试与调试高频问答

import pytest
from unittest.mock import patch, MagicMock
from datetime import datetime, timezone


# 1) 参数化测试：把数据与逻辑分离
@pytest.mark.parametrize(
    "amount, discount_rate, expected",
    [
        (100, 0.0, 100.0),
        (100, 0.1, 90.0),
        (99.99, 0.15, 84.99),
        (0, 0.5, 0.0),
    ],
)
def test_calculate_discount(amount, discount_rate, expected):
    from app.pricing import calculate_discount
    assert calculate_discount(amount, discount_rate) == pytest.approx(expected)


# 2) fixture + 依赖覆盖：分层复用测试资源
@pytest.fixture
def db_conn():
    """每个测试一个独立事务，结束后回滚，保证测试间零污染"""
    from app.db import engine
    conn = engine.connect()
    trans = conn.begin()
    yield conn
    trans.rollback()
    conn.close()


@pytest.fixture
def mock_payment_gateway():
    with patch("app.services.payment_gateway.charge") as mock:
        mock.return_value = {"status": "success", "txn_id": "txn_123"}
        yield mock


# 3) 依赖注入让测试变简单（不要在函数内部硬编码）
def create_order(user_id: int, payment=None, db=None) -> dict:
    payment = payment or default_payment_gateway
    db = db or default_db
    txn = payment.charge(user_id, 99.0)
    order = db.insert_order(user_id=user_id, txn_id=txn["txn_id"])
    return order


def test_create_order_with_mock(db_conn, mock_payment_gateway):
    order = create_order(42, payment=mock_payment_gateway, db=db_conn)
    assert order["status"] == "created"
    mock_payment_gateway.assert_called_once_with(42, 99.0)


# 4) 冻结时间，消除 flaky
def test_token_expiry(monkeypatch):
    from app.auth import create_token, verify_token

    fake_now = datetime(2026, 1, 1, tzinfo=timezone.utc)
    monkeypatch.setattr("app.auth.datetime_now", lambda: fake_now)

    token = create_token(user_id=42, expire_seconds=3600)
    assert verify_token(token)["user_id"] == 42

    # 时间推进到 2 小时后，token 应过期
    monkeypatch.setattr(
        "app.auth.datetime_now",
        lambda: fake_now.replace(hour=fake_now.hour + 2),
    )
    from app.auth import TokenExpired
    with pytest.raises(TokenExpired):
        verify_token(token)


# 5) 异步测试
import pytest


@pytest.mark.asyncio
async def test_async_fetch():
    from app.clients import fetch_user
    with patch("app.clients.http_client.get",
               new=MagicMock(side_effect=lambda *a, **k: _fake_response())):
        user = await fetch_user(42)
    assert user["id"] == 42


async def _fake_response():
    class R:
        status_code = 200
        async def json(self):
            return {"id": 42}
    return R()


# 6) 契约测试：保证 API 变更不破坏下游
def test_api_contract():
    """用 schema 校验响应结构，防止破坏性变更上线"""
    from app.models import OrderSchema
    payload = {"id": 1, "amount": 99.0, "status": "paid", "created_at": "2026-01-01T00:00:00Z"}
    OrderSchema.model_validate(payload)         # 不抛异常即通过

    with pytest.raises(Exception):
        OrderSchema.model_validate({"id": 1})   # 缺字段应失败

三个月学习路线图

前面的题目告诉你「考什么」，这一节告诉你「怎么准备」。下面的路线图假设你每天能投入 2~3 小时、周末 4~5 小时，共 13 周。每周有一个可验证的产出物，没有产出物的学习等于没学。

总体节奏

第 1~4 周：打地基

第 5~8 周：上强度

第 9~11 周：做项目

第 12~13 周：冲面试

时间分配建议（在职 vs 在校）

项目清单与资料推荐

推荐项目（按难度递增）

必读资料（官方优先）

开源项目阅读清单

# 推荐按「小 -> 大」的顺序读源码，每个都只用半天，只看入口与主链路

# 1) 300 行级：搞清楚原理
- https://github.com/psf/requests      —— 了解 requests 的 Session 与适配器机制
- https://github.com/pallets/click      —— 了解命令行的设计模式（参数如何转成函数调用）
- https://github.com/samuelcolvin/pydantic —— 了解类型注解驱动的运行时校验

# 2) 3000 行级：学习工程组织
- https://github.com/encode/starlette    —— 理解 ASGI 的最小实现（本章前面已手写过一个）
- https://github.com/pallets/flask      —— 看 WSGI 的中间件链条如何串起来
- https://github.com/psf/black          —— 代码格式化与 AST 改写的工程实践

# 3) 大型项目：学设计
- https://github.com/django/django      —— 看框架级的中间件、信号、ORM 设计
- https://github.com/sqlalchemy/sqlalchemy —— ORM 的 Session、Unit of Work、Identity Map
- https://github.com/celery/celery      —— 分布式任务队列的完整设计

# 阅读方法（推荐）
#   1. 先跑起来，看它解决什么问题
#   2. 从 CLI 入口往下追三层，看主链路
#   3. 挑一个你不懂的点，问 AI 或查文档，然后回去读那一段
#   4. 写下「我理解的设计取舍」，这是面试最好的素材

刷题与面经资源的使用方法

1. 按专题分类刷，而不是随机刷。本章前面列的高频题型就是最好的分类：数据结构 20%、并发 15%、数据库 20%、Web 与工程 20%、算法其余 25%。
2. 先想 10 分钟再查答案。看答案的学习效率约为独立思考的 30%。
3. 每道题写下复杂度（时间、空间）与一句「如果输入规模翻 10 倍会怎样」。
4. 面经按目录学，同一题在不同公司可能问法不同，但考点相同。遇到不会的记下来，24 小时内补课。
5. 每 5 道错题整理一次，错因分为：概念不清 / 没想到边界 / 手写不熟 / 语言特性没掌握。四类错因的补救方式完全不同。

常见反模式与临场应对

面试前一天的三件事

1. 重读自己的两个项目：把架构图默画一遍，确认每个技术选型都能说清「为什么不用另一个方案」。
2. 翻错题本与卡壳本：只过标记为「必须会」的 20 题。
3. 准备好反问面试官的两个问题，例如：「团队目前的代码质量与部署流程是怎么做的？」「这个岗位半年内最期望达成的目标是什么？」——这会显著提升面试官的印象分。

本章小结

• 答题框架：复述确认 → 一句话结论 → 分点展开 → 反向验证 → 主动补刀；不会时展示推理路径而不是硬编。
• 语言基础：推导式与生成器的惰性差异、可变默认参数的求值时机、拷贝的深浅规则、GIL 的历史与现状、装饰器的定义时执行、MRO 与 C3 线性化、__eq__ 与 __hash__ 的绑定规则。
• 数据结构：list/tuple/set/dict 的底层与复杂度、deque 的单调队列、lru_cache 原理、稳定性、字典保序的版本演进。
• 并发：线程/进程/协程的选型标准由「CPU 密集 vs IO 密集」决定；事件循环里不能调用同步阻塞库；死锁四个必要条件与五种规避方式；进程池必须用可 pickle 的顶层函数。
• 数据库：四种隔离级别与幻读、索引失效场景、深分页的游标方案、连接池超时是最常见的「数据库慢」信号、Redis 分布式锁必须用 Lua 保证比对与删除的原子性、缓存一致性的「先更新 DB 再删缓存」。
• Web：WSGI 与 ASGI 的本质差异、四个必备中间件（请求 ID、错误处理、限流、幂等）、状态码与错误响应结构、测试的四个高频追问。
• 学习路线：13 周四阶段（打地基 → 上强度 → 做项目 → 冲面试），每阶段有可验证产出物；每天 2~3 小时在周末补足 4~5 小时。
• 资料：官方文档优先，Fluent Python 补原理，DDIA 攻高级系统设计，开源源码按小到大读三级。
• 反模式：背答案、只学不用、假装精通、刷题不写码、项目讲不透——每一种都有明确的补救动作。
• 面试本质：不是考你知道多少，而是考你能否被信任把事情做对。诚实标注能力边界，反而是最专业的表现。

练习题

1. 口述训练：GIL 专题：用不超过 3 分钟讲清「什么是 GIL、它锁的是什么、为什么 CPython 要有它、它带来了什么后果、有哪些缓解手段」。然后用一段代码验证你的说法：分别用线程与进程做 CPU 密集任务，打印耗时对比。
   参考答案要点必须包含：GIL 是 CPython 的实现细节而非语言规范；它保证同一时刻只有一个线程执行字节码；原因是 CPython 用引用计数管理内存，非原子操作需互斥保护；后果是纯 CPU 密集的多线程无法并行，但 IO 等待期间会释放所以 IO 密集仍可并发；缓解手段按优先级为——多进程、multiprocessing、释放 GIL 的 C 扩展（NumPy 的大运算）、3.13+ 的自由线程构建、以及把瓶颈下推到数据库/Redis。验证代码见本章 s1 第 4 题的 bench()。
2. 手写题：从零实现（不许用内置函数）：① 一个单线程安全的 LRU 缓存类；② 一个支持超时与重试的装饰器，必须保留原函数 __name__ 与 __doc__；③ 一个能对任意可迭代对象做分块（chunk）且支持惰性求值的生成器。要求：每题写测试用例。
   参考答案要点LRU 用 collections.OrderedDict + move_to_end + popitem(last=False)，或用 dict（3.7+ 保序）+ 双端链表手写；装饰器用 functools.wraps；chunk 生成器写成 def chunked(it, size): it = iter(it); while (chunk := list(itertools.islice(it, size))): yield chunk；测试至少覆盖：容量淘汰顺序、缓存命中计数、超时后异常抛出、__name__ 保持、最后一个不满块的边界。
3. 缓存系统设计题：为「商品详情」设计缓存层，要求：① 说明缓存穿透/击穿/雪崩的成因与对策；② 给出缓存与数据库的一致性保证方案并指出其代价；③ 给出热点 key 的保护策略；④ 说明 Redis 内存超限时的降级路径。写出可运行的核心代码。
   参考答案要点穿透——查不存在的 key：用布隆过滤器前置拦截 + 空值短 TTL 缓存；击穿——热点 key 过期：single-flight（Redis SET NX 锁）或逻辑过期不真正删 key；雪崩——大批 key 同时过期：TTL 加随机抖动 + 多级缓存 + 服务降级返回兜底数据。一致性：Cache-Aside「先更新 DB，再删除缓存」，代价是删除失败需要重试队列，可加延迟双删应对主从延迟。热点保护：本地缓存 + 互斥重建 + 永不过期逻辑 key。内存超限：maxmemory-policy allkeys-lru 淘汰 + 命中率骤降告警 + 直接回源（要有数据库保护：限流/熔断）。
4. SQL 与索引实战：给出一张 orders(id, user_id, status, amount, created_at)（100 万行），写 SQL 并分析：① 查询某用户近 30 天已支付订单（走什么索引）；② 按金额分页且每页 20 条（如何避免深分页）；③ 统计每个状态的数量（能否走索引，为什么）；④ 删除 3 年前的数据（如何不锁表）。
   参考答案要点①建 idx_user_time (user_id, created_at) 走索引，且 created_at > ? 作为范围条件正好命中联合索引第二列；②深分页用游标：WHERE (created_at, id) < (?, ?) ORDER BY created_at DESC, id DESC LIMIT 20，并建 (created_at, id) 索引；③SELECT status, count(*) FROM orders GROUP BY status 在有 status 索引时可用索引扫描（覆盖），MySQL 5.7+ 与 PG 支持，但全表 count 无意义，属于「能走索引但不该走」；④分批删除：DELETE FROM orders WHERE created_at < '2023-01-01' LIMIT 5000 循环执行，或用 WHERE id IN (SELECT id ... LIMIT 5000)，避免一次锁百万行导致长时间阻塞与大量 undo。
5. 并发场景设计题：一个「秒杀」接口要支撑 1 万并发下单，库存 100 件。请给出技术方案：数据库层面怎么保证不超卖？Redis 层面怎么优化？用户侧怎么防刷？并说明每一步的取舍。
   参考答案要点数据库层最终兜底：UPDATE stock SET available = available - 1 WHERE sku_id = ? AND available > 0 靠行锁与条件判断保证不超卖，或用 SELECT ... FOR UPDATE 悲观锁（吞吐低），更常用「Redis 预扣 + MQ 异步下单 + 定时对账补偿」。Redis 层：DECR 原子扣减（Lua 脚本保证判断与扣减原子），热点 key 前置到本地缓存或做 key 分散（stock_1..stock_9 分桶降低单 key 压力）；下单资格用 Redis Set 的 SISMEMBER + SADD 的原子操作（Lua）限制一人一单。用户防刷：设备指纹 + 手机号 + IP 三维度限流、验证码/滑块、请求签名防伪造、行为验证。取舍说明：纯 Redis 方案快但有一致性风险（扣了库存消息丢了），必须设计「库存预扣 + 超时未支付回补 + 定时对账」的补偿链路；纯数据库方案简单可靠但吞吐有限（数千 QPS 就吃力）。
6. 项目包装题：选你做过的（或本章练习做过的）一个项目，按「背景 → 方案 → 取舍 → 结果 → 反思」五段式写一份 800 字的项目描述，其中「取舍」必须包含至少 3 个具体的技术选型对比（例如「为什么用 Redis 而不是直接查库」「为什么用异步任务而不是同步处理」「为什么不用微服务」）。
   参考答案要点结构模板：
   ①背景——业务痛点与量级（QPS、数据规模、团队规模）；
   ②方案——架构分层与关键技术；
   ③取舍——3 个「A vs B vs C」的决策，每个都要说出当时的选择理由 + 现在的回看（哪怕有回看也承认「当时考虑不周，现在会改成 X」）；
   ④结果——量化指标（延迟从 X 降到 Y、QPS 从 A 到 B、故障率下降）；
   ⑤反思——踩过的坑与改进（这一段最能体现成长，面试官最爱听）。
   写作提示：每个技术名词后面跟一句「它解决了什么问题」，否则听起来像在背技术栈。
7. 源码阅读报告：读 requests 源码的 sessions.py 与 adapters.py（或 starlette 的 applications.py），回答：① Session 相比直接调 requests.get 多了什么能力？② 连接池是如何复用的？③ 重试是怎么实现的？输出一份 500 字报告并画出调用链。
   参考答案要点Session 的能力：默认配置持久化（headers/params/hooks/auth 跨请求继承）、Cookie 自动管理与跨域保持、连接池复用（底层 HTTPAdapter 持有 urllib3.PoolManager）、默认的 RedirectMixin 重定向处理、更方便的上下文管理（with Session() as s: 会自动 close()）。连接池复用：HTTPAdapter.send 调用 poolmanager.connection_from_url(url)，相同 host:port:scheme 得到同一个连接池；requests.get 每次新建 Session 与 PoolManager，因此无法复用连接（这是性能问题的常见来源）。重试：urllib3.util.retry.Retry 对象配置在 adapter 上，支持按状态码或连接错误重试、指数退避、通过 mount() 挂到 Session 上。调用链：Session.request → send（准备 request）→ adapter.send → urllib3 → socket 层。
8. 综合实战：三个月计划的个人版本：根据你当前的真实水平（第几章、每天能投入多少小时、目标岗位是什么），写出一份属于自己的 13 周计划。要求：① 每周有明确产出物；② 每周包含算法题数量与面经题数量；③ 第 11 周必须有一个可用 Docker 一键部署的项目；④ 计划要包含 2 个「失败预案」（某周没完成时怎么补）。
   参考答案要点计划模板示例（在职转行版）：
   W1-2 每天 2 小时学语法+数据结构，周末 5 小时做「命令行待办 + 通讯录」；每天 10 道简单题。
   W3-4 函数/OOP/异常，每天 3 道中等题，周末做「成绩管理系统 + 补 100 道题」。
   W5-6 标准库+并发，每天 2 道中等题，周末做「并发爬虫」，对比三种并发模型耗时。
   W7-8 数据库+Web，手写 30 条 SQL，周末做「博客 API + 鉴权」。
   W9-10 测试与性能，给已有项目补 pytest 覆盖 80%，做一次 py-spy 定位实战。
   W11 Docker+Nginx+CI/CD，把博客 API 容器化并写好部署文档。
   W12-13 面经冲刺 + 简历 + 3 次模拟面试。
   失败预案示例：若 W6 未完成并发爬虫，则把 W7 的数据库学习减半顺延，优先保证 W11 的可部署项目完成（因为它是简历上最硬的证据）。每周日晚上做 15 分钟的进度复盘，只调整下周计划，不修改整体目标。