进阶篇 / 第 15 章

第 15 章 · 并发编程：线程、进程与异步

并发、并行与 GIL：先建立正确的心智模型

1.1 两个词的区别

中文里常把「并发」和「并行」混用，但它们在技术上严格不同：

并发的关键收益是「等待时间被隐藏」：一个程序 90% 的时间在等 I/O（网络、磁盘、数据库），那么重叠等待就能接近拿到线性加速。而并行带来的是「同一时刻干更多活」，对 CPU 密集型计算是刚需。

1.2 GIL 到底是什么

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器内部的一把大锁。同一时刻只允许一个线程执行 Python 字节码。它存在的历史原因是保护解释器内部状态（早期 Python 2 的引用计数在多线程下会出错），而不是为了限制你写多线程。

因此，CPU 密集任务在 CPython 里用多线程是负优化——线程越多，切换开销越大：

# cpu_bound.py —— 演示 CPU 密集任务下线程反而更慢
import time
import threading


def heavy(n: int) -> int:
    """纯计算：反复自增直到达到 n，制造 CPU 负载。"""
    x = 0
    for _ in range(n):
        x += 1
    return x


def run_sequential(n: int) -> float:
    t0 = time.perf_counter()
    heavy(n); heavy(n); heavy(n); heavy(n)
    return time.perf_counter() - t0


def run_threads(n: int, count: int = 4) -> float:
    threads = [threading.Thread(target=heavy, args=(n,)) for _ in range(count)]
    t0 = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - t0


if __name__ == "__main__":
    N = 5_000_000
    seq = run_sequential(N)
    thr = run_threads(N)
    print(f"串行 4 任务: {seq:.3f}s")
    print(f"线程 4 任务: {thr:.3f}s  （通常比串行慢！）")

1.3 GIL 的边界：什么时候会被释放

GIL 不是全程持有。解释器在以下情况会主动释放它：

1. 执行 I/O 系统调用（socket.recv、file.read、time.sleep）时；
2. 执行字节码的时间片用完（默认 sys.setswitchinterval()，历史上是 5ms）；
3. 执行某些 C 扩展主动释放 GIL 并声明自己线程安全（如 NumPy 的大部分 ufunc、hashlib、压缩库、数据库驱动、ssl）；
4. Python 3.13 起提供可配置 GIL（Py_GIL_DISABLED），3.14 起自由线程（free-threaded）构建已进入官方版本，但默认仍是带 GIL 的构建。

threading：线程基础

2.1 创建与启动线程

Python 3.3 起推荐使用 concurrent.futures 或 threading.Thread；threading.start_new_thread() 这种老 API 已经不推荐使用。

# threading_basic.py
import threading
import time
from datetime import datetime


def worker(name: str, times: int, lock: threading.Lock) -> None:
    """工作线程：每次打印一条带锁的日志，模拟真实业务。"""
    for i in range(times):
        # 临界区：多个线程共享资源时必须加锁，否则日志可能串行交错成乱码
        with lock:
            print(f"[{datetime.now():%H:%M:%S.%f}] {name} 第 {i} 次")
        time.sleep(0.05)          # I/O 或等待，故意放开 CPU


def main() -> None:
    lock = threading.Lock()
    threads = [
        threading.Thread(target=worker, args=("线程A", 3, lock), name="A"),
        threading.Thread(target=worker, args=("线程B", 3, lock), name="B"),
    ]

    print("主线程：等待所有子线程结束")
    for t in threads:
        t.start()                 # start() 才是真正启动
    for t in threads:
        t.join()                  # 等它结束，阻塞主线程

    print("主线程：全部结束，当前线程名 =", threading.current_thread().name)
    print("活动线程数 =", threading.active_count())


if __name__ == "__main__":
    main()

预期输出类似：

主线程：等待所有子线程结束
[10:21:33.120481] 线程A 第 0 次
[10:21:33.120536] 线程B 第 0 次
...
主线程：全部结束，当前线程名 = MainThread
活动线程数 = 1

2.2 常用 API 速查

2.3 线程安全的字典与计数器

Python 自带的无锁容器在 CPython 里靠 GIL 恰好不出错，但这不是语言层面的保证（一旦自由线程构建普及，多个操作之间的原子性就不再成立）。生产代码请用专门实现：

# thread_safe_containers.py
import threading
from concurrent.futures import ThreadPoolExecutor


def count_primes(limit: int) -> int:
    """统计 2..limit 以内的素数个数（CPU 计算 + 每步故意让出）。"""
    total = 0
    for n in range(2, limit):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            total += 1
    return total


def main() -> None:
    # 1) 手工加锁的共享计数器
    lock = threading.Lock()
    counter = {"n": 0}

    def bump() -> None:
        for _ in range(1000):
            with lock:
                counter["n"] += 1

    with ThreadPoolExecutor(max_workers=4) as pool:
        list(pool.map(lambda _: bump(), range(4)))
    print("共享计数器 =", counter["n"])          # 4000

    # 2) 线程安全的队列（queue.Queue 内部自带锁）
    import queue
    q: queue.Queue[int] = queue.Queue()
    for i in range(5):
        q.put(i)
    results = [q.get() for _ in range(q.qsize())]
    print("队列取出 =", results)                   # [0, 1, 2, 3, 4]

    # 3) 真正的并发（I/O 型）
    with ThreadPoolExecutor(max_workers=4) as pool:
        nums = list(pool.map(count_primes, [500, 800, 1200, 1600]))
    print("各线程素数计数 =", nums)


if __name__ == "__main__":
    main()

2.4 竞态条件的经典演示

# race.py —— 演示不加锁的 read-modify-write 竞态
import threading

counter = 0
LOSSY = 200_000


def bump() -> None:
    global counter
    for _ in range(LOSSY):
        counter += 1          # 等价于：tmp = counter; tmp += 1; counter = tmp


def main() -> None:
    ts = [threading.Thread(target=bump) for _ in range(4)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print("期望 =", LOSSY * 4)
    print("实际 =", counter)   # 通常小于期望值，丢失了更新
    print("丢失 =", LOSSY * 4 - counter)


if __name__ == "__main__":
    main()

预期输出（具体数字随机，但 实际 < 期望 的规律稳定）：

期望 = 800000
实际 = 799996
丢失 = 4

queue 与线程池：真正工程里最常用的组合

3.1 queue 模块

queue 是线程间安全传递数据的标准通道，包含：Queue（FIFO）、LifoQueue（栈）、PriorityQueue（按优先级）、SimpleQueue（最简）。

# queue_demo.py —— 生产者-消费者
import queue
import threading
import time
from dataclasses import dataclass, field


@dataclass(order=True)
class Task:
    """按优先级出队的任务。"""
    priority: int
    name: str = field(compare=False)


def producer(q: queue.Queue, count: int) -> None:
    for i in range(count):
        time.sleep(0.01)                    # 模拟慢速生产
        q.put(Task(priority=count - i, name=f"任务{i}"))
    q.put(None)                             # 毒丸（poison pill）：结束信号


def consumer(q: queue.Queue, wid: int) -> None:
    while True:
        task = q.get()
        try:
            if task is None:
                return                       # 收到毒丸，本消费者退出
            time.sleep(0.02)
            print(f"消费者{wid} 正在处理 {task.name}（优先级 {task.priority}）")
        finally:
            q.task_done()                   # 必须调用，否则 join() 会死等


def main() -> None:
    q: queue.Queue = queue.Queue(maxsize=10)  # 有界队列 -> 背压
    ps = [threading.Thread(target=producer, args=(q, 5)) for _ in range(2)]
    cs = [threading.Thread(target=consumer, args=(q, i)) for i in range(3)]

    for t in ps + cs:
        t.start()
    q.join()                                 # 等待所有 task_done()
    print("全部任务处理完毕，队列大小 =", q.qsize())


if __name__ == "__main__":
    main()

3.2 ThreadPoolExecutor

手写 Thread 加 Queue 很快会变得繁琐。concurrent.futures.ThreadPoolExecutor（线程池执行器）把「线程创建 + 复用 + 任务提交 + 结果回收 + 异常封装」全做好了。

# pool_basics.py —— 线程池四种提交方式
import time
from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
    wait,
    FIRST_COMPLETED,
)


def fetch(url: str, delay: float) -> dict:
    """模拟一次网络请求。"""
    time.sleep(delay)
    return {"url": url, "bytes": len(url) * 1024, "delay": delay}


def main() -> None:
    urls = ["a.example", "b.example", "c.example", "d.example", "e.example"]

    with ThreadPoolExecutor(max_workers=3, thread_name_prefix="pool") as pool:
        # 1) submit：逐个提交，拿到 Future 对象
        futures = {pool.submit(fetch, u, 0.2): u for u in urls}

        # 2) 阻塞等待全部完成，再按提交顺序取结果
        for fut in futures:
            print("按提交顺序:", fut.result()["url"])

        # 3) as_completed：谁先完成先处理（更常用）
        futures2 = [pool.submit(fetch, u, 0.1) for u in urls]
        for fut in as_completed(futures2):
            print("按完成顺序:", fut.result()["url"])

        # 4) 异常不会在 submit 处抛出，只在 result() 处抛出
        pool.submit(fetch, "x.example", -1)   # delay 负数 -> 内部异常
        # 用 wait() 观察状态
        done, not_done = wait(futures2, return_when=FIRST_COMPLETED)
        print(f"已完成 {len(done)} 个，未完成 {len(not_done)} 个")


if __name__ == "__main__":
    main()

Future 的关键 API：

3.3 超时、取消与优雅停机

# pool_timeout.py —— 超时不等于中断（重要事实）
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout


def slow(x: int) -> int:
    time.sleep(2)
    return x * 2


def main() -> None:
    with ThreadPoolExecutor(max_workers=2) as pool:
        fut = pool.submit(slow, 21)
        try:
            print(fut.result(timeout=0.3))
        except FutureTimeout:
            print("等待超时！")

        # 关键事实：线程无法被强制杀死，它仍在后台跑完 2 秒
        print("主线程可以继续做别的事……")
        time.sleep(2.2)
        print("回头再看结果:", fut.result(), "done =", fut.done())


if __name__ == "__main__":
    main()

等待超时！
主线程可以继续做别的事……
回头再看结果: 42 done = True

# graceful_shutdown.py —— 优雅停机 + 协作式取消
import threading
import time
from concurrent.futures import ThreadPoolExecutor


class Stopper:
    """把取消信号投递给所有工作线程。"""

    def __init__(self) -> None:
        self.stop = threading.Event()

    def should_stop(self) -> bool:
        return self.stop.is_set()


def worker(idx: int, stopper: Stopper, out: list[int]) -> None:
    for i in range(100):
        if stopper.should_stop():      # 每个循环检查取消信号
            print(f"worker {idx} 收到取消信号，提前退出")
            return
        out.append(idx * 100 + i)
        time.sleep(0.01)


def main() -> None:
    stopper = Stopper()
    out: list[int] = []
    with ThreadPoolExecutor(max_workers=4) as pool:
        futs = [pool.submit(worker, i, stopper, out) for i in range(4)]
        time.sleep(0.15)
        stopper.stop.set()              # 广播取消
        for f in futs:
            f.result()
    print("已完成工作单元数 =", len(out))


if __name__ == "__main__":
    main()

进程退出的规则：with ThreadPoolExecutor(...) as pool: 退出时自动调用 shutdown(wait=True)，会等待已提交任务完成。也可以显式 pool.shutdown(wait=True, cancel_futures=True) 取消未启动的任务。

3.4 模拟真实场景：并发抓取多个接口

# fetch_api.py —— 线程池 + 超时 + 失败重试 + 结果聚合
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass


@dataclass
class Result:
    name: str
    ok: bool
    data: dict | None = None
    error: str | None = None


class RateLimitedClient:
    """带最大重试次数的假客户端。"""

    def __init__(self, max_retry: int = 3) -> None:
        self.max_retry = max_retry
        self._attempts: dict[str, int] = {}

    def get(self, name: str) -> dict:
        self._attempts[name] = self._attempts.get(name, 0) + 1
        # 模拟：前两次失败，第三次成功
        if self._attempts[name] < 3 and not name.endswith("_always_fail"):
            raise ConnectionError(f"{name} 第 {self._attempts[name]} 次失败")
        if name.endswith("_always_fail"):
            raise TimeoutError(f"{name} 永久超时")
        return {"id": name, "value": len(name) * 10}


def fetch_one(client: RateLimitedClient, name: str) -> Result:
    last: Exception | None = None
    for attempt in range(1, client.max_retry + 1):
        try:
            time.sleep(0.05)                     # 假装网络延迟
            return Result(name=name, ok=True, data=client.get(name))
        except (ConnectionError, TimeoutError) as exc:
            last = exc
            time.sleep(0.02 * attempt)            # 指数退避
    return Result(name=name, ok=False, error=str(last))


def main() -> None:
    client = RateLimitedClient()
    targets = ["users", "orders", "items", "always_fail"]
    results: list[Result] = []

    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=4) as pool:
        futs = {pool.submit(fetch_one, client, t): t for t in targets}
        for fut in as_completed(futs):
            results.append(fut.result())
    cost = time.perf_counter() - t0

    for r in sorted(results, key=lambda r: r.name):
        status = "OK " if r.ok else "ERR"
        print(f"[{status}] {r.name:14s} -> {r.data or r.error}")
    print(f"总耗时 {cost:.3f}s（若串行需 ~{(len(targets)) * 0.21:.3f}s）")


if __name__ == "__main__":
    main()

[ERR] always_fail    -> items永久超时
[OK ] items           -> {'id': 'items', 'value': 50}
[OK ] orders          -> {'id': 'orders', 'value': 60}
[OK ] users           -> {'id': 'users', 'value': 50}
总耗时 0.216s（若串行需 ~0.840s）

信号量、限流与线程本地存储

4.1 Semaphore 限流

# semaphore_limit.py
import threading
import time

sem = threading.Semaphore(2)   # 全局同一把锁，最多 2 个线程同时进入


def task(i: int) -> None:
    with sem:                     # 进入临界区前获取令牌
        print(f"任务{i} 开始（并发数 <= 2）")
        time.sleep(0.1)
        print(f"任务{i} 结束")


ts = [threading.Thread(target=task, args=(i,)) for i in range(5)]
for t in ts:
    t.start()
for t in ts:
    t.join()
print("全部完成")

4.2 线程本地存储 threading.local

# tls.py
import threading

local = threading.local()


def worker() -> None:
    local.value = threading.current_thread().name
    local.count = 0
    for _ in range(3):
        local.count += 1
        print(f"{local.value}: count={local.count}")


ts = [threading.Thread(target=worker, name=f"T{i}", daemon=False) for i in range(2)]
for t in ts:
    t.start()
for t in ts:
    t.join()
print("主线程看不到子线程的属性:", hasattr(local, "value"))

4.3 Event / Condition / Barrier 速用

# sync_primitives.py
import threading
import time

# 1) Event：主线程发令枪
start_evt = threading.Event()
done_evt = threading.Event()


def waiter(idx: int) -> None:
    start_evt.wait()             # 阻塞直到 set()
    print(f"线程{idx} 收到发令")
    time.sleep(0.05)
    done_evt.set()


# 2) Barrier：等齐人数再一起出发
barrier = threading.Barrier(3)


def phase_worker(idx: int) -> None:
    barrier.wait()               # 等待所有 3 个线程到达
    print(f"线程{idx} 通过栅栏，进入下一阶段")


# 3) Condition：缓冲区非空才唤醒消费者
cv = threading.Condition()
buf: list[int] = []


def cond_producer() -> None:
    for i in range(3):
        with cv:
            buf.append(i)
            cv.notify()         # 唤醒一个等待者
        time.sleep(0.02)


def cond_consumer() -> None:
    got = 0
    while got < 3:
        with cv:
            while not buf:      # 必须用 while 循环检查，防止虚假唤醒
                cv.wait()
            buf.pop()
            got += 1
    print("消费者收到", got, "个元素")


ts = [threading.Thread(target=waiter, args=(i,)) for i in range(2)]
for t in ts:
    t.start()
time.sleep(0.1)
start_evt.set()
done_evt.wait(timeout=1)
for t in ts:
    t.join()

for i in range(3):
    threading.Thread(target=phase_worker, args=(i,)).start()

p = threading.Thread(target=cond_producer)
c = threading.Thread(target=cond_consumer)
p.start(); c.start(); p.join(); c.join()

multiprocessing：绕过 GIL 的多进程

5.1 什么时候该上多进程

1. 任务是 CPU 密集（纯 Python 计算、图像/视频编解码、密码学、大规模数值运算）；
2. 需要真正利用多核（os.cpu_count() 颗核）；
3. 单个任务粒度足够大，能抵偿「进程创建 + 数据序列化」的固定开销；
4. 需要更强的故障隔离（某进程崩溃不影响主进程）——如 ProcessPoolExecutor 中单个任务抛 BrokenProcessPool 时可以整体重启。

5.2 最简示例

# mp_basic.py —— 必须 if __name__ == "__main__" 保护
import multiprocessing as mp
import os
import time


def cpu_task(n: int) -> tuple[int, str]:
    """纯计算，验证多进程真的跑在多个 CPU 上。"""
    x = 0
    for _ in range(n):
        x += 1
    return os.getpid(), f"done {x}"


def report(result: tuple[int, str]) -> None:
    pid, msg = result
    print(f"子进程 PID={pid} -> {msg}")


def main() -> None:
    ctx = mp.get_context("spawn")   # 显式选启动模型，见 5.4
    print("主进程 PID =", os.getpid(), "可用的 CPU =", os.cpu_count())

    with ctx.Pool(processes=4) as pool:
        results = pool.map(cpu_task, [3_000_000] * 4)
    for r in results:
        report(r)

    t0 = time.perf_counter()
    with ctx.Pool(processes=4) as pool:
        pool.map(cpu_task, [3_000_000] * 4)
    print(f"多进程耗时 {time.perf_counter() - t0:.3f}s")


if __name__ == "__main__":
    main()

预期输出：4 个任务的 PID 互不相同，说明确实分布到不同进程；耗时约等于串行的 1/4（4 核机器）。

5.3 通信：管道、队列、共享内存

# mp_comm.py
import multiprocessing as mp
from multiprocessing import Queue, Pipe


def writer(q: Queue, p_conn) -> None:
    for i in range(3):
        q.put(("queue", i))
    p_conn.send(("pipe", "hello"))
    p_conn.close()


def reader(q: Queue, c_conn) -> None:
    for _ in range(3):
        print("收到", q.get())
    print("收到", c_conn.recv())
    c_conn.close()


def main() -> None:
    ctx = mp.get_context("spawn")
    q: Queue = ctx.Queue()
    parent, child = ctx.Pipe(duplex=False)   # duplex=False -> 单向

    w = ctx.Process(target=writer, args=(q, child))
    r = ctx.Process(target=reader, args=(q, parent))
    w.start(); r.start()
    w.join(); r.join()
    print("子进程退出码:", w.exitcode, r.exitcode)


if __name__ == "__main__":
    main()

# shared_mem.py —— 大数组用共享内存，别用 Queue 传
import multiprocessing as mp
from multiprocessing import shared_memory
import numpy as np


def worker(name: str, size: int, doubled: mp.Array) -> None:
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray((size,), dtype=np.float64, buffer=shm.buf)
    arr[:] = np.arange(size)
    print("子进程算出的前 5 项:", arr[:5])
    shm.close()          # 关闭映射（不要 unlink）


def main() -> None:
    N = 10_000
    shm = shared_memory.SharedMemory(create=True, size=N * 8)
    buf = np.ndarray((N,), dtype=np.float64, buffer=shm.buf)
    buf[:] = 0

    # 最稳妥的双向方式其实是让子进程直接写主进程传入的数组
    ctx = mp.get_context("fork")
    proc = ctx.Process(target=worker, args=(shm.name, N, None))
    proc.start(); proc.join()

    print("主进程读回前 5 项:", buf[:5])
    print("总和 =", buf.sum())
    shm.close()
    shm.unlink()


if __name__ == "__main__":
    main()

5.4 启动模型：fork / spawn / forkserver

multiprocessing.set_start_method("spawn")          # 全局设置
ctx = multiprocessing.get_context("spawn")            # 局部推荐，推荐这个

5.5 ProcessPoolExecutor

# process_pool.py
import multiprocessing as mp
import time
from concurrent.futures import ProcessPoolExecutor, as_completed


def crunch(n: int) -> int:
    s = 0
    for i in range(1, n + 1):
        s += i * i          # 平方和
    return s


def maybe_fail(n: int) -> int:
    if n < 0:
        raise ValueError("n 不能为负")
    return crunch(n)


def main() -> None:
    ctx = mp.get_context("fork")   # Linux 下用它提速
    with ProcessPoolExecutor(max_workers=4, mp_context=ctx) as pool:
        t0 = time.perf_counter()
        futures = [pool.submit(crunch, i * 500_000) for i in range(1, 5)]
        for fut in as_completed(futures):
            print("结果:", fut.result())
        print(f"4 任务并行耗时 {time.perf_counter() - t0:.3f}s")

        # 异常隔离：某个子进程抛错不会打断其他任务
        fut_bad = pool.submit(maybe_fail, -1)
        try:
            fut_bad.result()
        except Exception as exc:
            print("捕获子进程异常:", type(exc).__name__, exc)


if __name__ == "__main__":
    main()

5.6 什么时候根本不用自己写多进程

# parallel_jobs.py —— 优先用进程池
import time
from concurrent.futures import ProcessPoolExecutor


def is_prime(n: int) -> bool:
    if n < 2:
        return False
    if n % 2 == 0:
        return n == 2
    for i in range(3, int(n ** 0.5) + 1, 2):
        if n % i == 0:
            return False
    return True


def count_in_range(lo: int, hi: int) -> int:
    return sum(1 for n in range(lo, hi) if is_prime(n))


def main() -> None:
    LIMIT = 1_000_000
    t0 = time.perf_counter()
    total = count_in_range(0, LIMIT)
    print(f"单进程: {total}, {time.perf_counter() - t0:.2f}s")

    t0 = time.perf_counter()
    chunks = [(i * LIMIT // 8, (i + 1) * LIMIT // 8) for i in range(8)]
    with ProcessPoolExecutor(max_workers=8) as pool:
        parts = list(pool.map(count_in_range, *zip(*chunks)))
    print(f"8 进程: {sum(parts)}, {time.perf_counter() - t0:.2f}s")


if __name__ == "__main__":
    main()

asyncio：协程与异步 I/O

6.1 事件循环的基本模型

asyncio 是 Python 的标准异步框架，核心概念：

# asyncio_intro.py —— 从串行到并发的三个版本
import asyncio
import time


# 版本 1：纯串行（同步写法）
def sync_work(name: str, delay: float) -> str:
    time.sleep(delay)
    return f"{name} 完成"


# 版本 2：协程串行 await（仍然是串行！）
async def awaited_sequentially() -> list[str]:
    out = []
    for name, d in [("A", 0.5), ("B", 0.5), ("C", 0.5)]:
        await asyncio.sleep(d)           # sleep 会让出控制权
        out.append(f"{name} 完成")
    return out


# 版本 3：asyncio.gather 真并发
async def gathered() -> list[str]:
    async def job(name: str, d: float) -> str:
        await asyncio.sleep(d)
        return f"{name} 完成"

    return list(await asyncio.gather(
        job("A", 0.5), job("B", 0.5), job("C", 0.5)
    ))


async def main() -> None:
    t0 = time.perf_counter()
    await awaited_sequentially()
    print(f"await 串行: {time.perf_counter() - t0:.2f}s")   # 约 1.50s

    t0 = time.perf_counter()
    print(await gathered())
    print(f"gather 并发: {time.perf_counter() - t0:.2f}s") # 约 0.50s


if __name__ == "__main__":
    asyncio.run(main())

await 串行: 1.50s
['A 完成', 'B 完成', 'C 完成']
gather 并发: 0.50s

6.2 asyncio.gather 与 TaskGroup

# gather_vs_taskgroup.py
import asyncio
import time


async def work(i: int, fail: bool = False) -> str:
    await asyncio.sleep(0.1 * i)
    if fail:
        raise RuntimeError(f"任务 {i} 故意失败")
    return f"任务{i}"


async def use_gather() -> None:
    try:
        res = await asyncio.gather(work(1), work(2), work(3, fail=True))
        print(res)
    except RuntimeError as exc:
        print("gather 直接抛出第一个异常:", exc)
    # 注意：其它任务不会被取消，会继续在后台跑完！


async def use_task_group() -> None:
    # Python 3.11+ 推荐：一个失败 -> 自动取消同组其它任务
    async with asyncio.TaskGroup() as tg:
        for i in (1, 2, 3):
            tg.create_task(work(i, fail=(i == 2)))


async def main() -> None:
    await use_gather()
    try:
        await use_task_group()
    except* RuntimeError as eg:
        print("TaskGroup 收集到异常:", [str(e) for e in eg.exceptions])


if __name__ == "__main__":
    asyncio.run(main())

6.3 超时与取消

# asyncio_timeout.py
import asyncio


async def slow() -> str:
    await asyncio.sleep(5)
    return "终于完成"


async def main() -> None:
    # 1) asyncio.timeout 上下文管理器（3.11+ 推荐）
    try:
        async with asyncio.timeout(0.2):
            await slow()
    except TimeoutError:
        print("超时！TimeoutError 被抛出")

    # 2) 旧的 wait_for
    try:
        await asyncio.wait_for(slow(), timeout=0.2)
    except asyncio.TimeoutError:
        print("wait_for 超时")

    # 3) 超时 + 优雅取消
    task = asyncio.create_task(slow())
    try:
        async with asyncio.timeout(0.2):
            await task
    except TimeoutError:
        print("任务已被取消，done =", task.done(),
              "cancelled =", task.cancelled())

    # 4) Shield：即使外部取消也要跑完
    async def important() -> str:
        await asyncio.sleep(0.1)
        return "关键结果"

    try:
        async with asyncio.timeout(0.05):
            await asyncio.shield(important())
    except TimeoutError:
        print("shield 内部仍会完成")


if __name__ == "__main__":
    asyncio.run(main())

超时！TimeoutError 被抛出
wait_for 超时
任务已被取消，done = True cancelled = True
shield 内部仍会完成

6.4 用 aiohttp 做真并发请求

标准库 urllib 不是异步的。要并发发 HTTP 请求需要异步客户端，如 aiohttp 或 httpx。下面的示例使用 httpx（第 16 章会详细讲）：

# async_fetch.py —— 需要 pip install httpx
import asyncio
import time

import httpx


URLS = [
    "https://httpbin.org/get",
    "https://httpbin.org/uuid",
    "https://httpbin.org/json",
    "https://httpbin.org/headers",
]


async def fetch_one(client: httpx.AsyncClient, url: str) -> tuple[str, int, float]:
    t0 = time.perf_counter()
    try:
        r = await client.get(url, timeout=5.0)
        return url, r.status_code, time.perf_counter() - t0
    except httpx.HTTPError as exc:
        return url, -1, time.perf_counter() - t0


async def main() -> None:
    async with httpx.AsyncClient(
        headers={"User-Agent": "python-tutorial/1.0"},
        limits=httpx.Limits(max_connections=10),
    ) as client:
        t0 = time.perf_counter()
        results = await asyncio.gather(
            *(fetch_one(client, u) for u in URLS),
            return_exceptions=True,
        )
        total = time.perf_counter() - t0

    for url, status, cost in results:
        print(f"{status:4d}  {cost:.3f}s  {url}")
    print(f"总耗时 {total:.3f}s（串行需 ~{sum(r[2] for r in results):.3f}s）")


if __name__ == "__main__":
    asyncio.run(main())

6.5 结构化并发与常见反模式

# anti_patterns.py —— 对照理解
import asyncio
import time


# 反模式 1：在协程里调用阻塞函数 —— 整个事件循环被卡住
async def bad_blocking() -> str:
    time.sleep(1.0)      # 阻塞！其它协程全部停摆
    return "done"


# 正确：把阻塞操作丢到线程池
async def good_blocking() -> str:
    return await asyncio.to_thread(time.sleep, 1.0)   # 3.9+


# 反模式 2：for 循环里逐个 await
async def bad_loop() -> list[str]:
    out = []
    for i in range(1, 4):
        await asyncio.sleep(0.3)       # 总共 0.9s
        out.append(f"job{i}")
    return out


# 正确：一次性创建任务
async def good_loop() -> list[str]:
    async def job(i: int) -> str:
        await asyncio.sleep(0.3)
        return f"job{i}"

    return list(await asyncio.gather(*(job(i) for i in range(1, 4))))


async def main() -> None:
    t0 = time.perf_counter()
    await bad_loop()
    print(f"bad_loop:  {time.perf_counter() - t0:.2f}s")

    t0 = time.perf_counter()
    print(await good_loop())
    print(f"good_loop: {time.perf_counter() - t0:.2f}s")


if __name__ == "__main__":
    asyncio.run(main())

# cpu_offload.py —— 在 asyncio 里跑 CPU 任务
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


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


async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=4) as pool:
        t0 = time.perf_counter()
        results = await asyncio.gather(*[
            loop.run_in_executor(pool, heavy, 2_000_000) for _ in range(4)
        ])
        print("结果:", results)
        print(f"耗时 {time.perf_counter() - t0:.2f}s")


if __name__ == "__main__":
    asyncio.run(main())

6.6 自己造并发原语

# async_primitives.py
import asyncio
from contextlib import asynccontextmanager


async def worker(i: int, sem: asyncio.Semaphore, log: list[str]) -> None:
    async with sem:                     # 异步信号量限流
        await asyncio.sleep(0.05)
        log.append(f"w{i}")


@asynccontextmanager
async def timer(name: str):
    """异步上下文管理器：模拟异步计时代码。"""
    loop = asyncio.get_running_loop()
    t0 = loop.time()
    try:
        yield
    finally:
        print(f"{name} 耗时 {loop.time() - t0:.3f}s")


async def main() -> None:
    sem = asyncio.Semaphore(2)
    log: list[str] = []
    async with timer("限流批处理"):
        await asyncio.gather(*(worker(i, sem, log) for i in range(6)))
    print("执行顺序:", log)          # 任意交错，但同时最多 2 个

    # 异步迭代器：as_completed 模式
    async def numbers():
        for i in range(3):
            await asyncio.sleep(0.01)
            yield i

    out = [n async for n in numbers()]
    print("异步迭代结果:", out)

    # 事件
    ev = asyncio.Event()
    asyncio.get_running_loop().call_later(0.05, ev.set)
    await ev.wait()
    print("事件已触发")


if __name__ == "__main__":
    asyncio.run(main())

线程、进程、协程：可复现的性能对比

7.1 三类 workload 的基准测试

# bench_concurrency.py —— 一份脚本，测出三种方案的真实差距
"""
运行：python bench_concurrency.py
依赖：无（纯标准库，用 time.sleep 模拟 I/O 等待）
"""
from __future__ import annotations

import asyncio
import multiprocessing as mp
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from dataclasses import dataclass


N_TASKS = 20
IO_DELAY = 0.10        # 模拟每次网络/磁盘等待 100ms


@dataclass
class Bench:
    label: str
    cost: float


# ---------- 1. 串行基线 ----------
def task_io(name: str) -> str:
    time.sleep(IO_DELAY)          # 阻塞等待
    return name


# ---------- 2. CPU 型 ----------
def task_cpu(n: int) -> float:
    x = 0.0
    for i in range(600_000):
        x += i ** 0.5
    return x


def bench_sequential() -> list[Bench]:
    out = []
    t0 = time.perf_counter()
    for i in range(N_TASKS):
        task_io(f"io-{i}")
    out.append(Bench("串行 / I/O", time.perf_counter() - t0))

    t0 = time.perf_counter()
    for _ in range(N_TASKS):
        task_cpu(0)
    out.append(Bench("串行 / CPU", time.perf_counter() - t0))
    return out


# ---------- 3. 线程池 ----------
def bench_threads() -> list[Bench]:
    out = []
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=20) as pool:
        list(pool.map(task_io, [f"io-{i}" for i in range(N_TASKS)]))
    out.append(Bench("线程池(20) / I/O", time.perf_counter() - t0))

    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(task_cpu, [0] * N_TASKS))
    out.append(Bench("线程池(8) / CPU", time.perf_counter() - t0))
    return out


# ---------- 4. 进程池 ----------
def bench_processes() -> list[Bench]:
    out = []
    ctx = mp.get_context("fork")
    t0 = time.perf_counter()
    with ProcessPoolExecutor(max_workers=20, mp_context=ctx) as pool:
        list(pool.map(task_io, [f"io-{i}" for i in range(N_TASKS)]))
    out.append(Bench("进程池(20) / I/O", time.perf_counter() - t0))

    t0 = time.perf_counter()
    with ProcessPoolExecutor(max_workers=8, mp_context=ctx) as pool:
        list(pool.map(task_cpu, [0] * N_TASKS))
    out.append(Bench("进程池(8) / CPU", time.perf_counter() - t0))
    return out


# ---------- 5. asyncio ----------
async def amain() -> list[Bench]:
    out = []
    t0 = time.perf_counter()
    await asyncio.gather(*(asyncio.to_thread(task_io, f"io-{i}")
                           for i in range(N_TASKS)))
    out.append(Bench("asyncio+线程 / I/O", time.perf_counter() - t0))

    async def acpu() -> None:
        return await asyncio.to_thread(task_cpu, 0)

    t0 = time.perf_counter()
    await asyncio.gather(*(acpu() for _ in range(N_TASKS)))
    out.append(Bench("asyncio+线程 / CPU", time.perf_counter() - t0))
    return out


def report(rows: list[Bench]) -> None:
    base = {r.label.split(" / ")[1]: r.cost for r in rows if r.label.startswith("串行")}
    print(f"{'方案':28s} {'耗时':>9s} {'相对串行':>10s}")
    print("-" * 52)
    for r in rows:
        b = base.get(r.label.split(" / ")[1], r.cost)
        print(f"{r.label:28s} {r.cost:8.3f}s {r.cost / b:9.2f}x")


if __name__ == "__main__":
    print(f"任务数={N_TASKS}，I/O 延迟={IO_DELAY}s，CPU 单元=60 万次 sqrt\n")
    rows: list[Bench] = []
    rows += bench_sequential()
    rows += bench_threads()
    rows += bench_processes()
    rows += asyncio.run(amain())
    report(rows)

典型输出（4 核机器，具体数值随硬件变化）：

任务数=20，I/O 延迟=0.1s，CPU 单元=60 万次 sqrt

方案                           耗时     相对串行
----------------------------------------------------
串行 / I/O                   2.008s      1.00x
线程池(20) / I/O             0.117s      0.06x
进程池(20) / I/O             0.142s      0.07x
asyncio+线程 / I/O           0.128s      0.06x
串行 / CPU                   4.031s      1.00x
线程池(8) / CPU              4.198s      1.04x
进程池(8) / CPU              1.043s      0.26x
asyncio+线程 / CPU           4.215s      1.05x

7.2 选型决策树

1. 任务是 I/O（网络请求、文件、数据库）？→ 优先 asyncio 协程（并发量上千、资源占用低）；若依赖库只提供同步接口、或代码量小，线程池同样够用。
2. 任务是 CPU 密集的纯 Python 计算？→ 进程池 / 多进程；或换 NumPy、Cython、Rust 扩展；或用 subprocess 调外部程序。
3. 既要并发 I/O 又要并行 CPU？→ asyncio 负责编排，run_in_executor(ProcessPoolExecutor, ...) 负责重活。
4. 需要真正的取消/超时？→ asyncio 有真取消；线程没有；进程有（terminate）。
5. 已有大量同步第三方库？→ 先上线程池（asyncio.to_thread 逐个包装），不要为了异步而异步。
6. 任务是 GUI / 长驻服务且要求代码简单？→ 线程。tkinter、PyQt 的事件循环都是单线程 + 事件回调。

实战：并发抓取 + 限流 + 失败重试的完整服务

下面把本章所有零件组装成一个可直接运行的「并发任务调度器」：I/O 用协程并发，CPU 用进程池，带信号量限流、超时、指数退避重试、优雅取消与结构化并发。

# task_scheduler.py —— 完整可运行的并发调度器
"""
运行：python task_scheduler.py
依赖：标准库
"""
from __future__ import annotations

import asyncio
import itertools
import random
import signal
import time
from dataclasses import dataclass, field
from typing import Any


# ---------------- 数据模型 ----------------
@dataclass(order=True)
class Job:
    priority: int                 # 数字越小越优先
    name: str = field(compare=False)
    cost: float = field(compare=False, default=0.1)   # 模拟耗时
    fail_rate: float = field(compare=False, default=0.0)
    payload: dict[str, Any] = field(compare=False, default_factory=dict)


@dataclass
class JobResult:
    name: str
    ok: bool
    attempts: int
    value: str | None = None
    error: str | None = None


# ---------------- CPU 密集型工作（跑在进程池） ----------------
def cpu_intensive(payload: dict[str, Any]) -> dict[str, Any]:
    """真正的 CPU 计算，用于演示 run_in_executor + 进程池。"""
    n: int = payload.get("n", 200_000)
    s = 0
    for i in range(n):
        s += (i * i) % 7919
    return {"checksum": s, "size": n}


# ---------------- 调度器 ----------------
class Scheduler:
    """一个支持限流、超时、重试、优雅取消的并发调度器。"""

    def __init__(
        self,
        concurrency: int = 8,
        max_retry: int = 3,
        io_timeout: float = 1.0,
    ) -> None:
        self.sem = asyncio.Semaphore(concurrency)
        self.max_retry = max_retry
        self.io_timeout = io_timeout
        self.loop = asyncio.get_running_loop()
        self.pool = ProcessPoolExecutor(max_workers=4)
        self.stopping = False

    # --- I/O 型任务（协程 + 信号量 + 超时 + 退避重试） ---
    async def run_io(self, job: Job) -> JobResult:
        async with self.sem:
            for attempt in range(1, self.max_retry + 1):
                if self.stopping:
                    return JobResult(job.name, False, attempt, error="调度器正在关闭")
                try:
                    async with asyncio.timeout(self.io_timeout):
                        await asyncio.sleep(job.cost)     # 假装网络请求
                        if random.random() < job.fail_rate:
                            raise ConnectionError("随机网络故障")
                    return JobResult(job.name, True, attempt,
                                     value=f"{job.name} 在 {job.cost:.2f}s 后成功")
                except (TimeoutError, ConnectionError) as exc:
                    if attempt == self.max_retry:
                        return JobResult(job.name, False, attempt, error=str(exc))
                    backoff = 0.02 * (2 ** attempt)        # 指数退避
                    await asyncio.sleep(backoff)

            return JobResult(job.name, False, self.max_retry, error="未知")

    # --- CPU 型任务（卸载到进程池） ---
    async def run_cpu(self, job: Job) -> JobResult:
        async with self.sem:
            try:
                fut = self.loop.run_in_executor(self.pool, cpu_intensive, job.payload)
                out = await asyncio.wait_for(fut, timeout=30)
                return JobResult(job.name, True, 1,
                                 value=f"checksum={out['checksum']}")
            except Exception as exc:                       # noqa: BLE001
                return JobResult(job.name, False, 1, error=f"{type(exc).__name__}: {exc}")

    async def submit(self, job: Job) -> JobResult:
        return await (self.run_cpu(job) if job.payload else self.run_io(job))

    def request_stop(self) -> None:
        self.stopping = True

    def close(self) -> None:
        self.pool.shutdown(wait=False, cancel_futures=True)


# ---------------- 演示 ----------------
def build_jobs() -> list[Job]:
    return [
        Job(1, "配置加载", cost=0.05),
        Job(2, "数据库查询", cost=0.12),
        Job(2, "缓存预热", cost=0.08),
        Job(3, "外部API-A", cost=0.20, fail_rate=0.6),
        Job(3, "外部API-B", cost=0.15, fail_rate=0.0),
        Job(4, "图片压缩", payload={"n": 300_000}),
        Job(4, "特征计算", payload={"n": 250_000}),
        Job(5, "报表生成", cost=0.25, fail_rate=0.9),
        Job(6, "邮件通知", cost=0.03),
        Job(6, "日志归档", cost=0.10),
    ]


async def main() -> None:
    random.seed(42)                       # 可复现
    jobs = build_jobs()
    sched = Scheduler(concurrency=5, max_retry=3, io_timeout=1.0)

    # Ctrl+C 时优雅停止
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, sched.request_stop)
        except NotImplementedError:       # Windows 不支持
            pass

    print(f"提交 {len(jobs)} 个任务，优先级从高到低调度\n")
    t0 = time.perf_counter()
    results: list[JobResult] = []

    # TaskGroup：结构化并发，一个失败不影响其他
    # （注意：本例我们捕获异常，所以不会有失败逃出）
    ordered = sorted(jobs)
    async with asyncio.TaskGroup() as tg:
        tasks = []
        for job in ordered:
            tasks.append((job.name, tg.create_task(sched.submit(job))))
            await asyncio.sleep(0)          # 让出一次，保持优先级顺序提交

    for name, task in tasks:
        results.append(task.result())

    total = time.perf_counter() - t0

    print(f"{'任务':12s} {'状态':6s} {'尝试':>4s}  详情")
    print("-" * 72)
    ok = 0
    for r in sorted(results, key=lambda r: [j.name for j in ordered].index(r.name)):
        status = "成功" if r.ok else "失败"
        ok += r.ok
        print(f"{r.name:12s} {status:6s} {r.attempts:4d}  {r.value or r.error}")

    print("-" * 72)
    print(f"成功 {ok}/{len(results)}，总耗时 {total:.3f}s")
    print(f"串行预计耗时 ≈ {sum(j.cost for j in jobs if not j.payload):.2f}s + CPU 部分")
    sched.close()


if __name__ == "__main__":
    asyncio.run(main())

输出示例（随机项因种子固定可复现）：

提交 10 个任务，优先级从高到低调度

任务           状态   尝试  详情
------------------------------------------------------------------------
配置加载       成功     1  配置加载 在 0.05s 后成功
数据库查询      成功     1  数据库查询 在 0.12s 后成功
缓存预热       成功     1  缓存预热 在 0.08s 后成功
外部API-B      成功     1  外部API-B 在 0.15s 后成功
图片压缩       成功     1  checksum=1046927
特征计算       成功     1  checksum=914912
外部API-A      成功     2  外部API-A 在 0.20s 后成功
邮件通知       成功     1  邮件通知 在 0.03s 后成功
日志归档       成功     1  日志归档 在 0.10s 后成功
报表生成       失败     3  随机网络故障
------------------------------------------------------------------------
成功 9/10，总耗时 0.312s
串行预计耗时 ≈ 0.73s + CPU 部分

调试、观测与常见陷阱清单

9.1 并发代码的调试手段

# watchdog.py —— 死锁自动取证
import faulthandler
import sys
import threading
import time

faulthandler.enable(file=sys.stderr)


def slow_and_deadlock() -> None:
    lock_a, lock_b = threading.Lock(), threading.Lock()
    lock_a.acquire()
    time.sleep(0.1)
    lock_b.acquire()      # 与另一个线程互相等待 -> 死锁
    lock_b.release()
    lock_a.release()


def main() -> None:
    faulthandler.dump_traceback_later(5, exit=True)
    t = threading.Thread(target=slow_and_deadlock, daemon=True)
    t.start()
    t.join()               # 永远不会返回，5 秒后 faulthandler 打印全部线程栈并退出
    print("不会走到这里")


if __name__ == "__main__":
    main()

9.2 陷阱清单（建议逐条对照自己的项目）

1. 共享可变状态：默认不要在并发单元间共享对象，用消息传递（Queue / 消息队列）替代。
2. 锁顺序不一致导致死锁：多个锁必须全局定序（如永远先 lock_a 再 lock_b），或用 threading.Condition + wait() 避免嵌套持锁。
3. 在线程里直接抛异常会静默丢失：用 ThreadPoolExecutor（异常在 result() 里）而不是裸 Thread。
4. 无界队列导致内存爆掉：Queue(maxsize=N) 提供背压。
5. 线程池大小不是越大越好：线程过多会耗尽内存与文件描述符；I/O 密集一般 = CPU 核数 × (1 + 平均等待时间/计算时间)。
6. 进程池创建开销大：复用同一个 pool，不要在循环里 with ProcessPoolExecutor()。
7. 跨进程传大对象：pickle 开销可能吃掉全部收益，改用共享内存或让子进程自己加载数据。
8. fork + 多线程不安全：Linux 上进程已经开了线程后再 fork，子进程可能继承死锁的锁。尽量在启动时先 fork 再开线程。
9. asyncio 里忘了 asyncio.run：程序直接结束，什么都没干。
10. 未设超时：任何网络/文件操作都应带超时参数。

# pool_sizing.py —— 线程数选择的经验公式演示
import os
import threading
import time

CPU = os.cpu_count() or 4
WAIT = 0.5        # 平均 I/O 等待秒数
COMPUTE = 0.1     # 平均 CPU 计算秒数


def formula() -> int:
    """I/O 密集型线程数经验公式：N_cpu * (1 + W/C)。"""
    n = CPU * (1 + WAIT / COMPUTE)
    return min(int(n + 0.5), 64)


def main() -> None:
    print(f"CPU 核数 = {CPU}")
    print(f"I/O 线程建议值 = {formula()}")
    print(f"CPU 密集线程建议值 = {CPU}（实际建议直接用进程）")
    print(f"RLIMIT_NOFILE = {threading.Timer.__init__.__doc__ is not None}")


if __name__ == "__main__":
    main()

本章小结

• 并发 ≠ 并行：并发是逻辑上的交错推进（协程、线程），并行是同一时刻真正同时执行（多进程、多核）。
• GIL 让 CPython 的纯 Python 计算无法在多线程下并行；I/O 与释放 GIL 的 C 扩展（NumPy、hashlib、zlib）不受影响；Python 3.13+ 提供可配置 GIL，3.14 起有自由线程构建，但默认仍是带 GIL 的。
• 线程：Thread(target=...) + start()/join()；同步原语 Lock / RLock / Event / Condition / Semaphore / Barrier / local；裸线程里的异常会静默丢失，务必用线程池。
• 队列：queue.Queue 是线程间传数据的安全通道；maxsize 提供背压；join() 必须配合每条路径上的 task_done()（放 finally）。
• 线程池：ThreadPoolExecutor 是首选；submit/map/as_completed 三种用法；result(timeout=) 超时不会中断线程；要中断请用协作式取消（Event）或换进程/协程。
• 进程：绕过 GIL 的唯一方案，但内存大、通信必须 pickle、有 kill 能力。必须写 if __name__ == "__main__":；Linux 用 fork 快，Windows/macOS 用 spawn 安全。函数必须可 pickle。
• 协程：单线程高并发，切换开销极小，可开上万并发。关键是 gather / TaskGroup 真正并发、asyncio.timeout 控制超时、to_thread / run_in_executor 承接阻塞与 CPU 工作。
• 性能实测：I/O 场景三种方案差距极小（瓶颈在等待），CPU 场景只有进程池有线性加速。
• 选型口诀：I/O 走协程（同步库用线程），CPU 走进程（或换 NumPy/外部程序），混合场景用协程编排 + 进程池干活，任何 I/O 都写超时，任何并发单元都不共享可变状态。
• 观测：faulthandler.dump_traceback_later 做死锁取证，py-spy dump 无侵入采样，PYTHONASYNCIODEBUG=1 查 asyncio 问题。

练习题

1. 并发计数实验（线程安全）：写一个程序，启动 8 个线程，每个线程对各自区间内的整数做「是否为素数」判断并累加到共享变量 counter。分别用「不加锁」「用 Lock」「用 queue.Queue 汇总」三种方式实现，比较结果正确性与耗时差异。

   参考答案要点：锁版本 global counter; with lock: counter += local_count——注意先在本地累加，再一次性加锁写入，把锁粒度降到最低；队列版本让每个线程 q.put(local_count)，主线程 sum(q.get() for _ in range(8))，完全无共享写。不加锁在 CPython 下可能「看起来对」（counter += 1 恰巧原子），但换成复合操作（如 counter.x += 1）立刻现形。

2. 线程池 API 演练：用 ThreadPoolExecutor(max_workers=3) 提交 10 个「随机 sleep 0.1~0.5 秒」的任务。分别用 map()、submit()+as_completed()、submit()+wait() 三种写法，记录每个任务的完成顺序与总耗时。再让其中 2 个任务抛异常，分别在 gather 式遍历（取 result()）下验证异常如何影响整体。

   参考答案要点：map 按提交顺序返回且遇到异常立即中断；as_completed 按实际完成顺序产出，单个任务异常在取 result() 时抛出但不影响其他任务，稳妥做法是 try/except 包住 fut.result()。wait(fs, timeout=3, return_when=ALL_COMPLETED) 返回 (done, not_done) 两个集合，未完成的 Future 调用 result() 会永久阻塞。

3. asyncio 语义对比：写三个协程版本（各 sleep 1 秒）：① for 循环里逐个 await；② gather；③ TaskGroup。分别计时并解释差异。然后让其中一个协程抛异常，对比 gather 与 TaskGroup 的行为差异（是否取消其它任务、异常以何种形式抛出）。

   参考答案要点：①约 3s；②约 1s；③约 1s。gather 默认「先异常先抛」但不取消兄弟任务；TaskGroup 捕获第一个异常后立即取消同组其余任务，最终抛出 ExceptionGroup（需用 except* 捕获）。这是 3.11+ 推荐写法。

4. 多进程与 GIL 验证：写一个 CPU 密集任务（如 800 万次浮点累加），分别用 ①串行 ②4 线程 ③4 进程 ④进程池 Pool(4).map 运行，各跑 3 轮取最小值。用 os.getpid() 打印每次执行所在的进程号，并用 psutil.Process().cpu_times()（或 resource.getrusage）统计自己进程占用的 CPU 时间，证明线程版 CPU 时间≈墙钟时间而进程版 CPU 时间≈墙钟×4。

   参考答案要点：线程版墙钟时间不降反升（切换开销）；进程版墙钟约降为 1/4。CPU 时间 = time.process_time() 只计本进程，线程版仍≈墙钟；验证多核请看系统级 CPU 占用（如 top 中该进程 >100%）。pool.map 要套 if __name__ == "__main__":。

5. 实现带退避重试的限流下载器（线程版）：实现 download(url)，用 queue.Queue(maxsize=20) 做生产者-消费者，4 个消费者线程；每条 URL 有 30% 概率随机失败；失败最多重试 3 次，退避时间 0.1 * 2**n；成功后把结果写入一个列表（用锁保护）。运行结束后输出成功率与总耗时。

   参考答案要点：q.put(url, timeout=1) 处理满队列；消费者 while True: url = q.get(); ...; q.task_done()，用毒丸 None 或 q.join() 判断结束；重试用 for attempt in range(3) + time.sleep(0.1 * 2 ** attempt)；务必 try/finally: q.task_done()，否则 join() 死锁。总耗时 ≈ 串行 / 4 加上退避累计。

6. asyncio + 进程池混合：用 asyncio.gather 同时提交 10 个任务，其中 5 个是 await asyncio.sleep() 的 I/O 型，5 个是通过 loop.run_in_executor(ProcessPoolExecutor, cpu_func, arg) 的 CPU 型。测量总耗时，并与「全部用 to_thread」对比，解释差异。若 CPU 任务数 > 进程数，说明为什么并发度不再提升。

   参考答案要点：I/O 部分 ≈ 最慢的那次 sleep，CPU 部分 ≈ 总计算量 / 进程数，总耗时 ≈ max(两者) + 调度开销。to_thread 版 CPU 部分退化（GIL）到接近串行。进程池大小 = N 时第 N+1 个任务必须排队；进程池复用很重要，每次新建 pool 的启动开销可达几十毫秒。

7. 死锁定位实验：构造一个必然死锁的程序（两个线程以相反顺序获取两把锁），用 faulthandler.dump_traceback_later(3, exit=True) 让它 3 秒后自动 dump 所有线程栈并退出，读懂输出中线程卡在哪一行。然后改造为固定加锁顺序（永远 A→B），验证不再死锁。

   参考答案要点：dump 输出中每个线程一段 Thread 0x... (most recent call first) 加文件行号，能直接看出两个线程分别停在 lock_b.acquire() 与 lock_a.acquire()。修复方式：①全局锁顺序约定；②用 try_acquire + 回退；③把锁换成一条消息队列串行化。

8. 选型报告（开放题）：针对下面 4 个场景，写出你的方案选择与理由，并指出最关键的一个参数：
   (a) 从 500 个 API 端点抓取 JSON，每个耗时约 200ms；
   (b) 对 2000 万行 CSV 做逐行聚合计数；
   (c) 一个长期运行的 Flask 服务，需要每分钟向 30 个上游发心跳；
   (d) 批量把 500 个 4K 视频转码为 H.264，机器有 16 核。

   参考答案要点：(a) 协程（或线程池），关键是连接池大小（如 50）与总超时；(b) 进程池，或干脆用 pandas 分块 read_csv(chunksize=) + 向量化，注意任务粒度要够大；(c) 常驻线程池（或 APScheduler + asyncio），别每分钟新建事件循环，关键是复用连接与优雅退出；(d) 外部程序：subprocess.run(["ffmpeg", ...]) 并行跑 8~16 个 ffmpeg 进程——转码本身就在 C 层，Python 只是调度器，绝不要用 Python 循环去转码。