第 3 周（9/21 一 – 9/27 日）：并发与网络——asyncio、httpx、测试进阶、性能¶

本周目标：pytest 进阶（fixture/mock/覆盖率）；线程/进程/subprocess 的选择；asyncio 结构化并发；httpx 同步/异步/流式/重试与 mock 测试；性能测量。中秋假期（9/25–27）三天 10h 交付小项目 并发抓取聚合器。

时长：周一–周四 4h × 4 + 周五–周日 10h × 3 = 46h。

版本基准（2026-09-05 核实）：httpx 0.28.x、tenacity 9.1.x、respx 0.23.x、pytest 9.1、pytest-asyncio（asyncio_mode = "auto"）。详见 research/10、research/12。

对应 learn-agent：M14+M15 → Day 3（并发调用多个模型、httpx.AsyncClient）与所有 API 调用；M13 subprocess → 让 agent 执行命令行工具；M12 → 工具与 agent 的测试。

周一 9/21（4h）｜M12 pytest 进阶¶

块 1（复习）：第 2 周周测错题；kb 项目遗留。

块 2（新知识）： | 知识点 | 档 | | --- | --- | | fixture：函数/模块/会话作用域、yield 清理、conftest.py 共享、fixture 依赖 fixture | A | | 内置 fixture：tmp_path、monkeypatch（环境变量/属性/sys.argv）、capsys、caplog | A | | parametrize 进阶：ids=、多参数、pytest.param(..., marks=pytest.mark.xfail)；自定义 marker + -m；pytest -x --lf -k 关键字 -q | A | | pytest.raises(match=)、pytest.approx、pytest.warns | A | | unittest.mock.patch（打补丁到使用处而非定义处）、MagicMock、AsyncMock、spec=；pytest-mock 的 mocker | A | | 覆盖率：pytest --cov=包 --cov-report=term-missing，看漏测的行；覆盖率 ≠ 正确性 | A | | 测试金字塔：单元多、集成少、端到端更少；依赖注入让测试不需要 mock | A | | hypothesis 基于属性的测试 | C |

块 3（练习）：exercises/week3/w3_01_pytest_advanced/ —— 反向练习：给你一个完整的 inventory.py（库存管理：Inventory.add/remove/total_value/low_stock/export(path)，依赖一个 Clock 协议与一个 PriceService 协议），test_inventory.py 只有函数名与 docstring 的 TODO。你要写出 ≥12 个测试：fixture 提供 Inventory，monkeypatch/mocker 替换 PriceService，parametrize 覆盖边界，raises(match=) 覆盖错误，tmp_path 测导出，caplog 断言一条 WARNING。验收：uv run pytest exercises/week3/w3_01_pytest_advanced --cov=inventory --cov-report=term-missing 覆盖率 ≥95%。

周二 9/22（4h）｜M13 线程、进程、subprocess¶

块 2（新知识）： | 知识点 | 档 | | --- | --- | | GIL 与三类任务：I/O 密集（等网络/磁盘）→ 线程或 asyncio；CPU 密集 → 进程；调用外部程序 → subprocess | A | | concurrent.futures.ThreadPoolExecutor：map、submit + as_completed、max_workers 经验值；异常在 result() 时抛出 | A | | threading.Lock/Queue 保护共享状态；竞态演示（不加锁的计数器） | A | | ProcessPoolExecutor：Windows 用 spawn，必须 if __name__ == "__main__":；参数必须可 pickle | A | | subprocess.run(cmd_list, capture_output=True, text=True, encoding="utf-8", timeout=, check=)；shell=False；CalledProcessError/TimeoutExpired；Windows 上 ["python", "-c", ...] 用 sys.executable | A | | 3.14 free-threading（无 GIL 构建）现状：可选构建，默认仍有 GIL；concurrent.interpreters（3.14） | C | | asyncio.to_thread 是明天把阻塞调用接进 asyncio 的桥 | 预告 |

练习：w3_02_threads_processes_subprocess.py | 函数 | 行为 / 测试 | | --- | --- | | fetch_all_threaded(items, fetch, max_workers=8) -> dict | 用线程池并发调用注入的 fetch(item)（测试里 fetch 是 time.sleep(0.05) 的桩），保持 item→结果 映射；总耗时 < 顺序耗时的一半 | | safe_counter(n_threads, increments) -> int | 多线程累加，用 Lock 保证结果正确（先写不加锁版本观察错误） | | count_primes_parallel(ranges) -> int | ProcessPoolExecutor 分段数素数；模块级函数；测试在 __main__ 守卫下也能 import | | run_command(cmd: list[str], timeout=10) -> CommandResult | dataclass returncode/stdout/stderr/duration；超时 → CommandTimeout（自定义）；非零退出不抛，由调用方决定 | | python_version_via_subprocess() -> str | 用 sys.executable -c 拿版本，练 capture_output+encoding | | producer_consumer_threads(items) -> list | queue.Queue + 哨兵值结束 |

周三 9/23（4h）｜M14a asyncio 基础¶

块 2（新知识，全部 A 档）： - 协程函数 async def 调用得到协程对象，不执行；await 交出控制权；asyncio.run(main()) 是唯一入口。 - Task：asyncio.create_task 立刻调度；await task 取结果；未 await 的 task 可能被垃圾回收——保存引用或用 TaskGroup。 - asyncio.gather(*coros) vs asyncio.TaskGroup（3.11+，结构化：一个失败全部取消，异常聚合为 ExceptionGroup）——新代码优先 TaskGroup。 - asyncio.sleep 模拟 I/O；在协程里调用 time.sleep/requests/大计算会卡死整个事件循环——用 await asyncio.to_thread(blocking_fn)。 - 顺序 vs 并发的直观实验：10 个 sleep(0.2) 顺序 2s、并发 0.2s。 - 测试：pytest-asyncio + asyncio_mode = "auto"（已在 pyproject.toml 配好）→ 直接 async def test_x():。

预测：

import asyncio, time
async def work(i):
    await asyncio.sleep(0.1); return i * 2
async def main():
    t0 = time.perf_counter()
    results = await asyncio.gather(*(work(i) for i in range(5)))
    print(results, round(time.perf_counter() - t0, 1))   # 0.1 还是 0.5？
    c = work(1); print(type(c)); c.close()                 # 协程对象未 await 会警告
asyncio.run(main())

练习：w3_03_asyncio_basics.py（测试是 async def）：async fetch_fake(key, delay)；async gather_all(keys) -> list（gather）；async run_group(keys) -> list（TaskGroup，结果按输入顺序）；async first_n_done(coros, n)（asyncio.as_completed）；async with_thread(fn, *args)（to_thread 包装阻塞函数，测试用 time.sleep）；async sequential_vs_concurrent(n, delay) -> tuple[float, float] 返回两种耗时（测试断言并发 < 顺序/2）。

周四 9/24（4h）｜M16 性能与观测基础 + asyncio 复习¶

块 2（新知识）： | 知识点 | 档 | | --- | --- | | 先测量再优化：time.perf_counter、timeit（命令行 python -m timeit）、cProfile + pstats.sort_stats("cumulative")、tracemalloc 峰值 | A | | 容器复杂度直觉：list 的 in 是 O(n)，set/dict 是 O(1)；字符串拼接 vs "".join；循环里重复 re.compile；functools.cache | A | | 3.15 将新增 profiling 采样分析器（Tachyon）；py-spy；scalene | C | | asyncio 复习：把周三练习的 run_group 改成限制并发数（预习 Semaphore） | A |

练习：w3_04_profiling.py：给出 4 个故意写慢的函数（列表去重用 in、循环拼接大字符串、每次循环 re.compile、重复计算斐波那契），要求写出 fast_ 版本通过同一组正确性测试，并用 benchmark(fn, *args, repeat=5) -> float 证明加速（测试只断言 fast 不慢于 slow，避免机器差异导致抖动）；profile_top(fn, n=5) -> list[str] 返回 cProfile 前 n 行函数名；peak_memory(fn) -> int 用 tracemalloc。

周五 9/25（中秋，10h）｜M14b asyncio 进阶¶

上午 3 块： | 知识点 | 档 | | --- | --- | | 超时：async with asyncio.timeout(3):（3.11+）；wait_for 老写法；超时后任务被取消 | A | | 取消：task.cancel()、CancelledError 必须重新抛出（不要吞）、finally 清理、asyncio.shield | A | | 限流：asyncio.Semaphore(n) 包住每个请求；速率限制的简单令牌桶 | A | | 队列：asyncio.Queue 生产者/消费者、join()、哨兵/task_done | A | | TaskGroup 中一个失败 → 其余取消 → ExceptionGroup；except* 分类处理 | A | | asyncio.Event/Lock/Condition；asyncio.Lock 何时需要（协程之间也有竞态：两次 await 之间状态可能变） | B |

下午 3 块： | 知识点 | 档 | | --- | --- | | 异步生成器 async def ... yield + async for；async with（异步上下文管理器）；contextlib.asynccontextmanager | A | | 同步与异步世界的桥：asyncio.to_thread（阻塞→异步）、asyncio.run（异步→同步，只能顶层调用一次）；在已运行的循环里再 asyncio.run 会报错 | A | | 调试：asyncio.run(main(), debug=True)、PYTHONASYNCIODEBUG=1、3.14 的 python -m asyncio ps <pid> / pstree 查看任务树；"coroutine was never awaited" 警告的含义 | B | | anyio/trio：httpx、FastAPI/Starlette、MCP SDK 底层用 anyio；学会 asyncio 即可读懂，暂不学 anyio API | C | | 常见错误：忘 await、在协程里 time.sleep、共享可变状态、gather 吞掉部分失败（return_exceptions=True 的取舍）、在 finally 里 await 已取消的任务 | A |

练习：w3_05_asyncio_advanced.py：async gather_with_limit(coros, limit) -> list；async fetch_all_with_timeout(fetch, keys, timeout) -> dict[key, result | None]（超时项为 None，其余正常返回）；async producer_consumer(n_items, n_workers) -> list；async def aticker(interval, count) 异步生成器；async retry_async(fn, attempts=3, backoff=0.01)（指数退避，最后一次抛出）；async first_completed(coros)（其余取消）；@asynccontextmanager async timed_section(label, sink: list)；async run_group_collect_errors(coros) -> tuple[list, list[Exception]]（TaskGroup + except*）。

晚上 2 块：读 research/10 第三节 30 分钟；重构第 2 周 kb 的 import 命令为并发读取文件（to_thread 或线程池）；写日志。

周六 9/26（中秋，10h）｜M15 httpx + 项目启动¶

上午 3 块 — httpx（A 档）： - httpx.Client() / AsyncClient() 复用连接（with/async with），不要每次请求新建；base_url、headers、params。 - Timeout(connect=5, read=30, ...) 显式设置；不设超时是生产事故的常见来源。 - 错误层次：RequestError（网络层）vs HTTPStatusError（raise_for_status() 后的 4xx/5xx）；response.json()、.text、.status_code。 - 流式：client.stream("GET", url) + aiter_lines()/aiter_bytes()；大文件下载写盘。 - 重试：幂等请求才重试（GET）；httpx.HTTPTransport(retries=3) 只重试连接错误；业务级重试用 tenacity：@retry(stop=stop_after_attempt(3), wait=wait_exponential(...), retry=retry_if_exception_type(...))，含异步版。 - 代理与环境变量（HTTP_PROXY/trust_env）——国内网络场景；HTTP/2（http2=True，需 h2）知道。 - 测试不打真网：httpx.MockTransport(handler) 或 respx；本周练习全部离线。 - httpx-sse（SSE 客户端）：知道存在，第 4 周服务端做 SSE 时对照。

下午 3 块 — 练习 w3_06_httpx.py（全部用 MockTransport，无网络）：make_client(base_url, timeout) -> httpx.Client；get_json(client, path) -> dict（raise_for_status）；async fetch_many(aclient, paths, limit) -> dict[path, dict | Exception]（Semaphore + TaskGroup，单个失败不影响其它）；async stream_lines(aclient, path) -> list[str]；@retry 包装的 get_json_retrying（对 503 重试，测试用计数 handler 让前两次 503）；download(client, path, dest: Path) -> int 分块写盘返回字节数；classify_error(exc) -> str（网络/超时/状态码）。

晚上 2 块 — 项目启动：读下节规格，画模块图，建 projects/aggregator/ 骨架并跑通 uv run aggregator --help（骨架已给）。

周日 9/27（中秋，10h）｜项目 + 周测¶

上午 3 块 + 下午 3 块：完成聚合器；晚上：周测（exercises/week3/quiz_week3.md，40 分钟）+ AI 评审 + 周日志；预览 research/11 核心结论。

小项目：并发抓取聚合器 projects/aggregator/（≈16h）¶

用途：给一批 URL，并发抓取 JSON，限流、超时、重试，输出汇总（JSON/CSV/终端表格），并能对比"顺序 / 线程 / asyncio"三种实现的耗时。为了不依赖外网，项目自带一个桩服务器（asyncio 写的极简 HTTP 服务，可配置延迟与失败率），演示、测试、基准全部打本地。

结构（骨架已给）：

projects/aggregator/
├── pyproject.toml           # [project.scripts] aggregator = "aggregator.cli:app"；依赖 httpx tenacity typer rich pydantic pydantic-settings
├── src/aggregator/
│   ├── settings.py          # 并发数、超时、重试次数、输出格式（pydantic-settings，前缀 AGG_）
│   ├── models.py            # FetchResult(url, status, elapsed_ms, data|None, error|None)（Pydantic）；Summary
│   ├── client.py            # make_async_client(settings)；fetch_one（超时 + tenacity 重试 + 错误分类）
│   ├── engine.py            # fetch_all_async（TaskGroup + Semaphore）；fetch_all_threaded；fetch_all_sequential
│   ├── report.py            # 汇总统计（成功率、p50/p95 耗时）、rich 表格、JSON/CSV 导出（原子写）
│   ├── stub_server.py       # 已完整：`aggregator serve-stub --port 8001 --delay 0.1 --fail-rate 0.2`，/items/{n} 返回 JSON
│   ├── logging_config.py
│   └── cli.py               # fetch / bench / serve-stub / version
└── tests/                   # MockTransport 版单元测试 + 一个用 stub server 的集成测试（会话级 fixture 起停服务器）

验收：uv run aggregator serve-stub 起服务后，uv run aggregator fetch --count 50 --limit 8 --timeout 2 输出表格与 summary.json；uv run aggregator bench --count 30 显示三种实现耗时，线程池比顺序快 ≥5 倍、asyncio 至少快 2 倍（本机 Windows 上 httpx 异步栈开销偏大，见 projects/aggregator/README.md 的排查记录；Linux 上两者接近）；失败率 20% 时成功率因重试提升；Ctrl+C 能干净退出（取消任务、关闭客户端）；pytest --cov=aggregator ≥80%；ruff/Pylance 零错误；≥8 次提交推送 cnb.cool。

本周阅读（详见 research/10、12、13）¶

本周陷阱清单（周测必考）¶

patch 打到定义处而不是使用处｜fixture 作用域写错导致状态泄漏｜覆盖率 100% 但没断言｜ProcessPoolExecutor 没有 __main__ 守卫（Windows 无限递归启动）｜线程共享列表不加锁｜subprocess 用 shell=True 拼字符串｜忘 await｜协程里 time.sleep｜未保存的 Task 被回收｜gather 里一个异常导致其余继续跑却拿不到结果｜超时后不处理 CancelledError｜吞掉 CancelledError｜每次请求新建 Client｜不设超时｜对 POST 盲目重试｜测试打真网｜asyncio.run 嵌套