14b AI 应用后端补课

第一部分 速记页

Redis

鉴权

MySQL 与稳定性

部署

第二部分 易混对照

第三部分 面试口述稿

3.1 “Redis 为什么快？它是单线程的吗？”

快主要有几个原因。第一，数据都在内存里，读写不碰磁盘。第二，执行命令的是一个主线程，命令一个接一个执行，不需要加锁，也没有线程切换的开销。第三，它用 I/O 多路复用，一个线程就能同时管理成千上万个连接，哪个连接有数据就处理哪个。第四，数据结构做了很多紧凑的实现，小的哈希、列表会用更省内存的编码。

说它单线程，准确地说是”执行命令是单线程”。Redis 6.0 以后，从网络读请求、往网络写结果这部分可以开多线程，但命令本身还是在主线程里顺序执行，所以单条命令天然是原子的。

这也带来一个要注意的点：一条慢命令会卡住所有人。比如对一个很大的集合做 KEYS *、HGETALL 一个几十万字段的哈希，执行期间别的命令都得等。所以线上禁止用 KEYS，要遍历用 SCAN，大键要拆开。

3.2 “缓存穿透、击穿、雪崩分别是什么，怎么解决？”

这三个都是缓存没挡住、请求打到数据库的情况，区别在原因。

穿透是查一个根本不存在的数据，缓存里没有，数据库里也没有，所以每次都穿过缓存打到数据库。有人故意用随机编号刷接口就是这种。办法是把”不存在”也缓存起来，存一个空值，过期时间设短一点，比如一分钟；数据量大的话可以前面放一个布隆过滤器，先判断编号可能不可能存在。

击穿是一个热点键过期的那一瞬间，大量并发请求同时发现缓存没了，全都去查数据库。办法是重建缓存时加互斥锁，只让一个请求去查数据库，其他请求稍等再读缓存；或者热点键干脆不设过期，由后台定时刷新。我做过实验，20 个并发请求打一个刚失效的键，不加锁数据库被查了 20 次，加锁后只查 1 次，总耗时几乎一样。

雪崩是大面积失效，比如一批键设了同样的过期时间，同时过期；或者 Redis 整个挂了。办法是过期时间加一个随机值把它们打散；Redis 做主从加哨兵或集群保证高可用；数据库前面做限流和降级，宁可部分请求失败也不能把数据库压垮。

3.3 “缓存和数据库怎么保持一致？”

我用的是最常见的旁路缓存模式。读的时候先查缓存，没有再查数据库，查到后写回缓存并设过期时间。写的时候先更新数据库，再删除缓存，而不是去改缓存。

为什么删而不是改：两个请求同时写，更新数据库的顺序和更新缓存的顺序可能反过来，最后缓存里留的是旧值。删掉的话，下一次读自然会加载最新值，逻辑简单得多。

为什么是先改数据库再删缓存，而不是先删缓存：先删缓存的话，删完到数据库更新完成之间，有读请求进来发现缓存为空，就会把数据库里的旧值读出来写回缓存，这个旧值会一直留到过期。先改数据库再删缓存也不是绝对安全，但出问题需要非常凑巧的时序，概率低得多。

要求更高的话，可以删缓存失败时放进重试队列；或者订阅数据库的 binlog，由单独的程序负责删缓存。无论哪种，都给缓存设一个过期时间兜底，保证最终一致。强一致的数据，比如余额，就别放缓存。

3.4 “用 Redis 怎么实现分布式锁？要注意什么？”

加锁用一条命令：SET 锁名 随机值 NX PX 过期毫秒数。NX 表示键不存在才设置成功，PX 同时设上过期时间。这两件事必须在一条命令里完成，如果先 SETNX 再 EXPIRE，中间进程崩了，锁就永远不会过期。

过期时间是为了防止持锁的进程崩了、锁永远不释放。但它带来另一个问题：任务如果比过期时间长，锁会提前失效，别人就能拿到锁。这时候如果原来的进程干完活直接 DEL，删掉的其实是别人的锁。所以值必须是每次加锁生成的随机值，释放时用一个 Lua 脚本先比对值是不是自己的，是才删。Lua 脚本在 Redis 里是原子执行的，比对和删除之间不会插进别的命令。我做过实验，直接 DEL 的版本，第三个客户端也拿到了锁，出现两个同时在跑；比对后再删的版本就不会。

任务时长不确定的话，加一个看门狗：后台线程每隔过期时间的三分之一，用同样比对值的方式续期，任务结束再停掉。Java 的 Redisson 就是这么做的。

还要知道它的局限：持锁进程如果卡顿了很久，比如长时间的垃圾回收，锁已经过期、别人已经拿到，它醒来后还会继续操作；Redis 主从切换时锁也可能丢。要求绝对不能并发的场景，要在被保护的资源那一侧校验，比如每次加锁拿一个递增编号，写数据库时带上，编号比已经写过的小就拒绝。

我的项目现在用的是进程内的 threading.Lock，只要 uvicorn 开两个 worker 进程就失效。我实验过，同时发 6 个请求，进程内锁放进来 2 个构建，换成 Redis 锁只放进来 1 个。

3.5 “限流有哪些算法？AI 应用的限流有什么特别？”

常见四种。固定窗口：按时间段计数，比如每 10 秒一个计数器，实现最简单，用 Redis 的 INCR 加过期时间就行，但边界有问题，在窗口结束前和下一个窗口开始后各打满一次，一两秒内能放过两倍的量，我实验里限额 5 次，边界前后各发 5 次，全放过去了。滑动窗口：用有序集合记下每次请求的时间戳，每次只数最近一个窗口内的，没有边界问题，同样的测试只放过 5 次，代价是每个请求都要存一条记录。令牌桶：按固定速度往桶里加令牌，请求拿到令牌才能过，允许一定的突发流量，长期平均速度受控，是网关里最常用的。漏桶：请求进队列按固定速度流出，输出最平滑。

多实例部署时计数要放在 Redis 里，而且”判断加计数”要原子，用 Lua 脚本写。

AI 应用的特别之处是，请求和请求的成本差别很大，一个请求可能花几百 token，也可能花几十万。所以除了按次数，还要按 token 用量做每个用户每天的配额。难点是调用之前不知道会花多少 token，常见做法是调用前检查剩余额度够不够一个预估值，调用完按实际用量扣减。超了返回 429，带上 Retry-After 头告诉客户端多久后再试。

3.6 “为什么要用任务队列？Redis 做队列会丢消息吗？”

Agent 任务一跑几分钟，不能在请求里同步执行。放在 Web 进程的后台线程里，像我项目现在这样，有两个问题：Web 进程一重启，正在跑的任务全没了；任务多了没法单独加机器。任务队列把”接收请求”和”执行任务”分开：接口只负责把任务写进队列、立刻返回任务编号，独立的工作进程从队列里取任务执行，状态写到 Redis 或数据库。

用 Redis 的 List 做队列，LPUSH 放、BRPOP 取，最简单，但取出来就从队列里删了，工作进程这时候崩了，任务就丢了。Redis 5 以后有 Streams，配合消费组用：消息被某个消费者读走后进入待确认列表，处理完调用 XACK 才算完成；消费者崩了，消息一直留在待确认列表里，别的消费者可以用 XAUTOCLAIM 把超时没确认的消息认领过来重新处理。我实验过这个流程，崩溃的那条任务被第二个工作进程认领后完成了。

这样的保证是”至少一次”：一条消息可能被处理两次，比如处理完、还没来得及确认就崩了。所以任务处理必须幂等。我的做法是处理完成后写一个”已完成”标记，处理前先检查。注意标记要在处理成功之后写，写在前面的话，处理到一半崩了，重试时会以为已经做完，任务就丢了。

实际项目里一般不自己写，Python 常用 Celery，异步项目可以用 arq，它们底层可以用 Redis 做中间件。消息量很大、需要严格顺序或者长期保存的，用 RabbitMQ 或 Kafka。

3.7 “Session 和 JWT 有什么区别？你会选哪个？”

Session 是服务端保存登录状态：登录成功后服务端生成一个会话记录，把会话编号放在 Cookie 里给浏览器，之后每次请求带上编号，服务端去查。好处是服务端完全掌控，想让某个用户下线，删掉记录就行。缺点是有状态，多台服务器要共享会话存储，一般放 Redis。

JWT 是把用户信息直接写进令牌里，由服务端用密钥签名。它分三段：头部、载荷、签名，用点隔开。服务端收到后只要验签，就知道令牌没被改过、是谁的、什么时候过期，不用查任何存储，天然适合多实例和服务之间调用。缺点是签发出去就很难提前作废，只能等它过期。

有两个常见误解要说清楚。第一，JWT 的载荷没有加密，只是 Base64URL 编码，谁拿到都能解开看，我实验里不用密钥就把载荷解出来了，所以里面不能放敏感信息。第二，解码时要指定允许的签名算法，否则可能被人用 alg: none 这类手法绕过验签。

选择上，面向浏览器的单体应用我倾向 Session 加 Redis，作废简单。需要给多个服务或者第三方调用，就用 JWT，访问令牌设短过期，比如 30 分钟，再配一个长一些的刷新令牌；需要强制下线的话，在 Redis 里维护一个作废名单，验签后再查一下。

3.8 “用户密码怎么存？API key 呢？”

密码绝对不能明文存，也不能用 MD5、SHA-256 直接算。这些哈希算得太快，数据库一旦泄露，攻击者用显卡每秒能试几十亿次，常见密码很快就被猜出来。要用专门为密码设计的、故意很慢而且吃内存的算法，OWASP 现在首推 Argon2id，最低配置是 19 MiB 内存、2 次迭代；其次是 scrypt、bcrypt。每个密码还要随机加盐，这样同一个密码两次哈希结果也不同，彩虹表就没用了。Argon2id 的结果字符串里自带算法参数和盐，验证时直接用。FastAPI 官方教程现在推荐的是 pwdlib 加 Argon2。

API key 不一样。它是系统生成的长随机字符串，熵很高，不存在”常见 key”可以猜，所以用 SHA-256 存哈希就够了，不需要慢哈希。流程是：生成后只在创建时显示一次，服务端只存哈希；请求来了把收到的 key 算一次哈希去查。一般还会加一个固定前缀，比如 sk-，方便代码扫描工具识别泄露。

另外，调大模型用的 API key 只能放在服务端的环境变量或密钥管理服务里，永远不能下发到前端。前端请求自己的后端，后端再去调模型，这样还能顺带做鉴权、限流和记账。

3.9 “线上有个接口变慢了，数据库这边你怎么排查？”

第一步是找到慢的 SQL。打开慢查询日志，设一个阈值，比如 long_query_time 设成 0.5 秒，超过的都会记下来，包括执行时间、扫描了多少行、返回了多少行。日志多的话用 pt-query-digest 按模式汇总，先处理总耗时最多的那几条。

第二步是对这条 SQL 做 EXPLAIN，主要看四列：type 是访问方式，ALL 就是全表扫描；key 是实际用了哪个索引，为空就是没用上；rows 是估计要扫描多少行；Extra 里如果有 Using filesort 说明要额外排序，Using temporary 说明用了临时表，Using index 则是好事，说明覆盖索引、不用回表。

第三步是对症处理。最常见的几种：没有索引就加，多条件查询建联合索引，把等值条件放前面、排序字段放后面；索引列上套了函数，比如 DATE(created_at) = '...'，改写成范围条件；联合索引跳过了最左列；深分页；应用代码里的 N+1 查询。

我做过一组实验，30 万行的工具调用表，按运行编号查一次运行的全部调用，没索引平均 57 毫秒，估计扫 29.5 万行；加上索引后 0.2 毫秒，扫 15 行。慢查询日志里 Rows_examined 和返回行数差距很大，就是典型信号。

3.10 “分页翻到很后面很慢，怎么办？”

LIMIT 280000, 20 这种写法，数据库并不能直接跳到第 28 万行，它要按顺序扫过前 28 万行再丢掉，越往后越慢。我实验里这条要 36.7 毫秒，慢查询日志显示扫描了 28 万零 20 行，只返回 20 行。

改法是游标分页：记住上一页最后一条的 id，下一页查 WHERE id > 上次的id ORDER BY id LIMIT 20，直接从主键索引定位，同样的位置只要 0.3 毫秒，而且不管翻到多深都一样快。代价是不能直接跳到第几页，只能上一页、下一页，但对话记录、运行历史这种按时间往下翻的列表正好合适。如果排序字段不唯一，比如按时间排序，就用”时间加 id”两个字段一起做游标。

必须支持跳页的话，可以先在覆盖索引上查出这一页的 id，再用这些 id 回表取完整数据，扫描的仍然很多，但每行代价小了。

3.11 “下游模型 API 不稳定，你的服务怎么保证可用？”

分几层。

最底层是超时和重试，02 篇讲过：每次调用必须有超时；网络错误、429、5xx 用指数退避加随机抖动重试，400、401、402 这种重试也没用的立刻失败。

第二层是熔断。如果模型服务已经持续出问题，每个请求还是要等到超时、重试几次，既拖慢用户，又给对方增加压力。熔断器统计连续失败次数，到了阈值就进入”打开”状态，这段时间内的请求直接失败或者走备用方案，根本不去调；过一段时间进入”半开”，放一个试探请求，成功就恢复正常，失败就继续打开。我实验里连续失败 3 次后打开，后面 3 个请求都没有打到主模型；主模型恢复后，一个试探请求成功就关闭了。

第三层是降级，也就是失败后走什么：切到提前评测过的备用模型，返回缓存的结果，或者关掉非核心功能，比如先不生成摘要。

第四层是舱壁，限制同时打到某个下游的并发数。比如用一个信号量，最多同时 3 个请求去调模型，其他的排队。这样即使模型变慢，也只是这几个位置被占住，不会把整个服务的连接和线程耗光。

3.12 “你的项目怎么部署？”

用 docker-compose 起三个服务：API、Redis、Nginx。

API 的镜像基于 python slim，Dockerfile 先复制依赖清单装依赖，再复制代码，这样改代码重新构建时不用重装依赖；容器里建一个普通用户运行，不用 root；.dockerignore 排除 .venv、.env 和数据库文件，防止密钥被打进镜像。

Redis 挂数据卷、开 AOF，重启后数据还在。compose 里给 Redis 和 API 都配了健康检查，API 用 depends_on 加 condition: service_healthy，等 Redis 真正能用了才启动；只写 depends_on 只能保证启动顺序，不保证已经可用。

Nginx 在最前面反向代理。SSE 的路径要注意三件事，我都实测过：一是关掉代理缓冲；二是不能对事件流开 gzip，我开了以后三条间隔一秒的事件被攒到最后一起到达；三是 proxy_read_timeout 时间内必须有数据，我把它设成 3 秒、后端 5 秒不发数据，连接就被断开了，加上每秒一次的心跳注释就能正常跑完。

还有一点，uvicorn 开多个 worker 就是多个进程，进程内的锁、内存缓存都不共享，所以并发控制必须放到 Redis 里。

3.13 “线上服务 CPU 或内存突然很高，你怎么查？”

先确认是哪个进程：top 或 htop 按 CPU、内存排序；容器环境用 docker stats。

CPU 高的话，Python 服务我会用 py-spy，它不用改代码、不用重启，py-spy dump --pid 进程号 能直接打出每个线程当前卡在哪一行，py-spy top 能看哪个函数最耗时。常见原因是死循环、正则回溯、在事件循环里做了 CPU 密集的计算。

内存高的话，先看是不是一直涨，一直涨通常是泄漏，比如全局字典、缓存没有上限，或者对话历史一直往里追加。dmesg 里有 Out of memory 就是被系统的 OOM killer 杀掉了。

其他常见的：ss -lntp 看端口被谁占用；lsof -p 进程号 看打开了多少文件和连接，Too many open files 就是连接没关或者 ulimit -n 太小；df -h 看磁盘是不是被日志写满了。

我在 C++ 项目里排查过线程和资源回收的问题，思路是一样的：先定位进程，再定位线程和代码位置，再看资源有没有正常释放。

第四部分 逐个详解

4.1 补课地图：哪些已经讲过

后端面试的知识点，有一大半已经在别的文章里讲过。先把它们列出来，这一篇不重复，只补空白。

把这些知识点放回一个上线的 AI 应用里，大概是下面这个样子。方框里标了本篇讲它的小节：

flowchart TB
    B[浏览器] -->|HTTPS / SSE| N["Nginx 反向代理<br/>4.10"]
    N --> A1["API 进程 1"]
    N --> A2["API 进程 2"]
    subgraph API["FastAPI（uvicorn 多 worker）"]
        A1
        A2
        AUTH["依赖注入做鉴权<br/>JWT、API key 4.7"]
    end
    API -->|"缓存 4.3 / 锁 4.4 / 限流 4.5"| R[("Redis")]
    API -->|"提交任务 4.6"| R
    R -->|"消费组取任务"| W["工作进程<br/>跑 Agent"]
    W -->|"事件写回流，SSE 续传"| R
    W -->|"熔断 / 降级 / 并发上限<br/>4.9"| LLM["模型 API"]
    API --> DB[("MySQL<br/>4.8")]
    W --> DB

4.2 Redis 基础

先说为什么会有 Redis。 一个 AI 应用后端里，有一类数据读写特别频繁，但每条都很小：某个用户这一分钟调了几次接口、验证码、登录状态、一次 Agent 运行走到第几步、模型回答的缓存。把它们放在哪儿，有三个直觉上的选择，各自都有问题：

• 放在 Python 进程里的字典：最快，但 4.1 那张图里 API 有好几个进程，还有单独的工作进程。每个进程有自己的一份字典，进程 1 记下”u1 已经调了 9 次”，进程 2 看不到。进程一重启，数据也没了
• 放在 MySQL：所有进程都能看到，但 MySQL 是为”数据不能丢、要支持复杂查询”设计的，每次写入要解析 SQL、走事务、写日志落盘。拿它来做”计数加一”这种事太重。而且同一行被高频更新时，行锁会让这些写入排队
• 放在 Memcached：2003 年为 LiveJournal 网站做的内存缓存服务，单独跑一个进程，所有程序通过网络连它，解决了”多进程共享”和”快”。但它的值只能是一段字符串。想在一个列表末尾加一条，得先把整个值读回来，在自己程序里改，再整个写回去。两个程序同时这么做，后写的会把先写的覆盖掉

Redis 就是在这个背景下出现的。2009 年，意大利程序员 Salvatore Sanfilippo 在做一个实时网站访问统计的产品，要给每个网站维护”最近 N 次访问”的列表，访问一多 MySQL 就跟不上了。他自己写了一个内存服务，思路和 Memcached 一样是独立进程、数据在内存、走网络访问，区别是值可以是列表、哈希这样的数据结构，而且”往列表里加一条""把字段加一”这种操作由 Redis 自己在内部完成，客户端只发一条命令，不用读出来再写回去。名字 Redis 是 REmote DIctionary Server 的缩写，意思就是”远程字典服务”：一个大家都能通过网络访问的字典。

flowchart LR
    subgraph S1["各存各的"]
        P1["API 进程 1<br/>dict: u1=9"]
        P2["API 进程 2<br/>dict: u1=3"]
    end
    subgraph S2["共用一份"]
        Q1["API 进程 1"] --> RS[("Redis<br/>u1=12")]
        Q2["API 进程 2"] --> RS
        Q3["工作进程"] --> RS
    end

四种做法放在一起比：

所以 Redis 在后端里的位置通常是：MySQL 负责存”不能丢、要查询”的正式数据，Redis 挡在前面，负责”要快、要多进程共享、丢一点能接受或能重建”的数据。它不是用来替代 MySQL 的。

顺带一句现状：Redis 原来是 BSD 开源协议，2024 年改成了限制云厂商商用的协议，Linux 基金会随后从旧代码分出了一个 Valkey，命令基本兼容；2025 年的 Redis 8 又加回了 AGPLv3 开源协议。面试一般不问这个，但云服务商的”Redis 兼容服务”很多底层已经是 Valkey，看到不用意外。

Redis：一个把数据放在内存里的键值数据库。每条数据有一个键（key，字符串）和一个值，值可以是好几种数据结构。读写都在内存里完成，单条命令一般在微秒级。

五种最常用的结构，以及在 AI 应用里的用法：

为什么快、单线程是什么意思：口述见 3.1。要记住的推论是：单条命令是原子的，但多条命令合起来不是。比如”先读计数、判断没超、再加一”如果分三条命令发，中间可能插进别人的命令。要把几步合成一个原子操作，用 Lua 脚本（4.4 节开始会反复用到）。

flowchart TB
    C1[客户端 1] --> MUX
    C2[客户端 2] --> MUX
    C3[客户端 N] --> MUX
    MUX["I/O 多路复用<br/>哪个连接有数据就读哪个<br/>（6.0 起读写网络可多线程）"] --> Q["命令排成一队"]
    Q --> T["主线程逐条执行<br/>单条命令天然原子"]
    T --> MEM[("内存里的数据")]

一条慢命令会让队伍后面所有命令一起等，这就是线上禁止 KEYS * 的原因。

过期：SET key value EX 秒数 或 EXPIRE key 秒数 给键设过期时间，TTL key 看还剩多少秒（-1 表示没设过期，-2 表示键不存在）。Redis 删过期键有两种方式：惰性删除，访问这个键时发现过期了才删；定期删除，后台每隔一会儿随机抽一批设了过期的键检查。所以过期的键在被删之前还会占一会儿内存。

淘汰：内存用到 maxmemory 上限时，按 maxmemory-policy 删键（官方文档）：

LRU 是近似的：Redis 不维护完整的访问顺序，而是随机抽几个键，删其中最久没用的。64 位系统上 maxmemory 默认是 0，也就是不限制，做缓存时一定要设。

持久化：Redis 数据在内存里，进程重启就没了，所以有两种落盘方式。RDB 每隔一段时间把整个数据集拍一张快照存成文件，文件小、恢复快，但两次快照之间的写入会丢。AOF 把每条写命令追加到文件里，默认每秒刷一次盘，最多丢约 1 秒的数据，文件会越来越大，Redis 会定期重写压缩。纯缓存可以都不开；存任务队列、锁、会话的，要开 AOF。

动手实验 1：五种结构各用一下。先装 Redis（macOS 上 brew install redis，然后 redis-server --port 6390 启动一个实例），再装 Python 客户端 pip install redis。保存为 redis_basics.py。

import time

import redis

r = redis.Redis(port=6390, decode_responses=True)
r.flushdb()

r.set("model:default", "deepseek-flash")
print("字符串:", r.get("model:default"))

r.set("verify:code:1001", "482913", ex=2)
print("剩余秒数:", r.ttl("verify:code:1001"))
time.sleep(2.1)
print("过期后:", r.get("verify:code:1001"))

r.hset("run:217", mapping={"status": "running", "step": 13, "tokens": 51234})
r.hincrby("run:217", "step", 1)
print("哈希:", r.hgetall("run:217"))

r.rpush("chat:u1", "你好", "帮我编译 zlib", "改成 aarch64")
r.ltrim("chat:u1", -2, -1)
print("列表只留最近两条:", r.lrange("chat:u1", 0, -1))

r.zadd("usage:today", {"u1": 52000, "u2": 8000, "u3": 130000})
print("有序集合 token 用量前两名:", r.zrevrange("usage:today", 0, 1, withscores=True))

print("计数器:", r.incr("req:count"), r.incr("req:count"))

实际输出（Redis 8.10.2，redis-py 8.1.0，Python 3.14）：

字符串: deepseek-flash
剩余秒数: 2
过期后: None
哈希: {'status': 'running', 'step': '14', 'tokens': '51234'}
列表只留最近两条: ['帮我编译 zlib', '改成 aarch64']
有序集合 token 用量前两名: [('u3', 130000.0), ('u1', 52000.0)]
计数器: 1 2

逐段解释：

• redis.Redis(port=6390, decode_responses=True)：创建一个客户端对象，连到本机 6390 端口。decode_responses=True 让返回值是 Python 字符串；不加的话返回的是字节串 b'...'
• Python 语法：关键字参数。port=6390 这种”名字=值”的写法叫关键字参数，按名字传，顺序无所谓，没传的用函数的默认值。r.set(..., ex=2) 里的 ex 就是”过期秒数”这个参数
• r.flushdb()：清空当前数据库，只在实验里用，线上绝对不要执行
• r.set("verify:code:1001", "482913", ex=2)：键名里的冒号只是命名习惯，把”类型:编号”分层，Redis 本身不关心
• r.hset("run:217", mapping={...})：mapping= 传一个字典，一次设置多个字段。Python 语法：字典 {"键": 值, ...}，按键取值
• 哈希里的 step 读回来是字符串 '14'：Redis 里存的全是字符串，数字要自己转
• r.ltrim("chat:u1", -2, -1)：只保留下标 -2 到 -1 的元素，也就是最后两个。负数下标从末尾往前数。这是”只保留最近 N 轮对话”的常用写法
• zrevrange(..., 0, 1, withscores=True)：按分数从高到低取第 0 到第 1 个，同时返回分数
• r.incr 是原子的：多个进程同时加一，不会丢

4.3 缓存：旁路模式和三种失效

先说为什么会有旁路缓存这种用法。 有了 Redis、Memcached 这种放在内存里的服务，接下来的问题是”数据库和缓存两份数据怎么配合”。一种想法是让缓存自己去连数据库，没有就自己查，但 Memcached 这类服务只会存取键值，不认识数据库。于是业界形成了一个分工：缓存只是个被动的存储，读的时候由应用自己先查缓存、没有再查库并写回去；写的时候由应用改库，再把缓存里那条删掉。 这个做法最有名的公开描述是 Facebook 2013 年在 NSDI 会议上的论文 Scaling Memcache at Facebook，他们把它叫作”按需填充的旁路缓存”（demand-filled look-aside cache），并写明写入时选择删除缓存而不是更新缓存，理由是删除操作重复执行多少次结果都一样。同一篇论文还专门讲了”一个热点键被频繁改动、大量读请求同时回源到数据库”的问题，他们的办法是给回源的请求发一个”租约”，同一个键 10 秒内只发一次，其余请求稍等再读缓存，思路和下面实验里的互斥锁重建一样。

穿透的常见解法里提到的布隆过滤器更早：1970 年 Burton H. Bloom 在论文里提出，用来在内存很小的前提下判断”某个东西肯定不在集合里”，当时的例子是 50 万个单词的断字词典，大部分单词可以用简单规则处理，只有少数要去磁盘查，先用它挡掉大部分不必要的磁盘访问。它会误判”在”，但不会误判”不在”，所以拿来挡”根本不存在的键”正合适。

**旁路缓存（Cache-Aside）**是最常用的缓存用法：

flowchart TB
    A[读请求] --> B{缓存里有吗}
    B -- 有 --> C[直接返回]
    B -- 没有 --> D[查数据库]
    D --> E[写回缓存并设过期时间]
    E --> C
    W[写请求] --> X[更新数据库] --> Y[删除缓存]

写的时候为什么删缓存而不是改缓存、为什么先改库再删缓存：口述见 3.3。

下图是”先删缓存、再改数据库”出错的时序。库里原来是旧值 v1，写请求要改成 v2：

sequenceDiagram
    participant W as 写请求
    participant C as Redis 缓存
    participant D as 数据库
    participant R as 读请求
    W->>C: 1. 删除缓存
    R->>C: 2. 读缓存，没有
    R->>D: 3. 读数据库，拿到旧值 v1
    W->>D: 4. 更新为 v2
    R->>C: 5. 把 v1 写回缓存
    Note over C,D: 缓存是 v1，数据库是 v2，直到过期前一直不一致

换成”先改数据库、再删缓存”，第 4 步在第 1 步之前完成，读请求读到的已经是 v2；要出错，需要读请求在更新前读到 v1、又在删除之后才写回，这个窗口小得多。

三种缓存没挡住的情况（口述见 3.2）：

动手实验 2：用一个”查项目配置”的函数模拟数据库，每次查询要 0.2 秒，并统计被查了几次。保存为 cache.py。

import json
import random
import threading
import time

import redis

r = redis.Redis(port=6390, decode_responses=True)
r.flushdb()

calls = {"n": 0}
calls_lock = threading.Lock()
KNOWN = {"zlib", "libpng", "curl"}


def load_from_db(name):
    with calls_lock:
        calls["n"] += 1
    time.sleep(0.2)
    if name not in KNOWN:
        return None
    return {"name": name, "build_system": "cmake"}


def get_project(name):
    key = f"project:{name}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)
    data = load_from_db(name)
    ttl = 600 + random.randint(0, 120)
    r.set(key, json.dumps(data), ex=ttl if data else 60)
    return data


def get_project_with_mutex(name):
    key = f"project:{name}"
    while True:
        cached = r.get(key)
        if cached is not None:
            return json.loads(cached)
        if r.set(f"rebuild:{key}", "1", nx=True, ex=5):
            try:
                data = load_from_db(name)
                r.set(key, json.dumps(data), ex=600 + random.randint(0, 120))
                return data
            finally:
                r.delete(f"rebuild:{key}")
        time.sleep(0.05)


def burst(fn, name, n=20):
    calls["n"] = 0
    r.delete(f"project:{name}")
    threads = [threading.Thread(target=fn, args=(name,)) for _ in range(n)]
    start = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return calls["n"], time.perf_counter() - start


calls["n"] = 0
get_project("zlib")
get_project("zlib")
print("旁路缓存：查两次 zlib，数据库被查", calls["n"], "次")

calls["n"] = 0
for _ in range(5):
    get_project("not-exist")
print("穿透防护：查 5 次不存在的项目，数据库被查", calls["n"], "次，缓存里存的是", r.get("project:not-exist"))

print("雪崩防护：zlib 的过期时间", r.ttl("project:zlib"), "秒（600 到 720 之间随机）")

n, cost = burst(get_project, "libpng")
print(f"击穿：热点键失效时 20 个并发请求，不加锁，数据库被查 {n} 次，耗时 {cost:.2f}s")
n, cost = burst(get_project_with_mutex, "libpng")
print(f"击穿：加互斥锁重建，数据库被查 {n} 次，耗时 {cost:.2f}s")

实际输出：

旁路缓存：查两次 zlib，数据库被查 1 次
穿透防护：查 5 次不存在的项目，数据库被查 1 次，缓存里存的是 null
雪崩防护：zlib 的过期时间 647 秒（600 到 720 之间随机）
击穿：热点键失效时 20 个并发请求，不加锁，数据库被查 20 次，耗时 0.22s
击穿：加互斥锁重建，数据库被查 1 次，耗时 0.23s

逐段解释：

• json.dumps(data) 把 Python 字典转成 JSON 字符串存进 Redis，json.loads 再转回来。Redis 只存字符串，结构化数据都要这样序列化
• json.dumps(None) 得到字符串 "null"，所以不存在的项目在缓存里存的是 null。下次 r.get 拿到的是 "null" 而不是 Python 的 None，就不会再去查库，这就是缓存空值。空值的过期时间只给 60 秒，防止这个项目后来被创建了却一直查不到
• ttl = 600 + random.randint(0, 120)：过期时间在 600 到 720 秒之间随机，同一批写进去的键不会同一秒过期，这就是防雪崩的随机过期
• Python 语法：f-string。f"project:{name}" 前面加 f，花括号里的变量会被替换成它的值。{cost:.2f} 表示保留两位小数
• Python 语法：条件表达式。ttl if data else 60 的意思是”如果 data 有值就用 ttl，否则用 60”
• get_project_with_mutex 是防击穿的写法：缓存没有时，先用 SET ... NX 抢一个”重建锁”，抢到的去查数据库并写回缓存；没抢到的睡 50 毫秒后重新读缓存。Python 语法：while True 是一直循环，直到里面 return 才退出
• Python 语法：try / finally。finally 里的代码不管 try 里是正常返回还是抛了异常都会执行，保证重建锁一定被删掉
• 重建锁也设了 5 秒过期：万一抢到锁的请求崩了，别人最多等 5 秒
• burst 同时启动 20 个线程去读同一个刚被删掉的键。Python 语法：列表推导式 [表达式 for _ in range(n)] 生成一个有 n 个元素的列表，_ 表示这个循环变量用不到。threading.Thread(target=fn, args=(name,)) 创建一个线程，target 是要执行的函数，args 是传给它的参数，注意只有一个参数时要写成 (name,)，逗号不能少，否则不是元组
• t.start() 启动线程，t.join() 等它结束
• calls_lock 是 Python 进程内的锁，只是为了让 20 个线程同时给计数加一时不出错，和 Redis 无关。Python 语法：with calls_lock: 进入时自动加锁，离开缩进块时自动释放

结果说明：不加锁时 20 个请求全部打到数据库；加锁后只有 1 个，而总耗时几乎一样（0.22 秒对 0.23 秒），因为其他 19 个请求只是多等了一下，数据库查一次也是 0.2 秒。热点数据量大、数据库查询慢时，这个差别就是数据库扛不扛得住。

AI 应用里缓存什么、不缓存什么：

• 适合缓存：embedding 结果（同一段文本向量不变）、检索结果、用户和项目配置、模型列表
• 小心缓存模型回答：只有输入完全相同、回答不依赖时间和用户、温度为 0 时才考虑，否则用户会拿到一模一样的”随机”结果（14 篇 4.8 节）
• 不要和提示词缓存混淆：那是模型服务商那边对相同前缀的计算做缓存，降低费用和延迟，不需要你用 Redis（06 篇）

4.4 分布式锁

为什么需要：threading.Lock 是进程内的锁，只对同一个 Python 进程里的线程有效。uvicorn 开了 --workers 2，就是两个独立进程，各有一把自己的锁，互相看不见；部署到两台机器更是如此。4.10 节的实验里，进程内锁在 2 个 worker 下放进来了 2 个构建。要让所有进程共用一把锁，锁就得放在大家都能访问的地方，Redis 最常用。

先说分布式锁是怎么来的。 最早认真做这件事的是大公司的基础设施团队。Google 的 Mike Burrows 2006 年在 OSDI 会议上发表了 Chubby，一个专门给内部系统提供锁的服务，用五台机器跑 Paxos 共识算法保证挂掉一两台锁也不会乱。Yahoo 的几位工程师随后做了开源的 ZooKeeper，2010 年在 USENIX ATC 发表了论文。这两个都可靠，但都是要单独部署、单独运维的一套集群。很多团队手上已经有 Redis，就想直接拿它加锁。早期的写法是 SETNX 再 EXPIRE 两条命令，下面表格里会讲它的毛病；Redis 2.6.12 起 SET 命令加了 NX、PX 这些选项（SET 命令文档），一条命令就能原子地”不存在才设置、同时设过期时间”，这才有了下面这套写法。Redis 作者后来又提出了多节点的 Redlock 算法，2016 年 2 月 Martin Kleppmann 写文章 How to do distributed locking 质疑它的安全性，Redis 作者第二天就发文回应（Is Redlock safe?），这场争论就是本节最后那张表里两个追问的出处。

加锁：SET lock:build <随机值> NX PX 30000

• NX：键不存在才设置，存在就失败，这就是”抢锁”
• PX 30000：30000 毫秒后自动过期，防止持锁的进程崩了锁永远不释放
• 随机值：每次加锁时新生成，证明”这把锁是我加的”

释放：不能直接 DEL，要先比对值。口述见 3.4。

完整的流程：

flowchart TB
    A{"SET lock 随机值<br/>NX PX 60000"} -- 失败 --> F[返回 409<br/>或稍后重试]
    A -- 成功 --> W["执行任务<br/>看门狗每 20 秒续期<br/>（值是自己的才续）"]
    W --> L{"任务结束<br/>Lua：值还是自己的吗"}
    L -- 是 --> DEL[删除锁]
    L -- 否 --> X["锁已过期被别人拿走<br/>什么都不删"]

sequenceDiagram
    participant A as 进程 A
    participant R as Redis
    participant B as 进程 B
    A->>R: SET lock A的值 NX PX 500
    Note over A: 任务比 500 毫秒长
    Note over R: 锁过期自动删除
    B->>R: SET lock B的值 NX PX 5000 → 成功
    A->>R: DEL lock（直接删）
    Note over R: B 的锁被 A 删掉了
    Note over A,B: 第三个进程又能拿到锁

动手实验 3：演示直接 DEL 的问题、比对后删除的正确写法，以及看门狗续期。保存为 lock.py。

import threading
import time
import uuid

import redis

r = redis.Redis(port=6390, decode_responses=True)
r.flushdb()

RELEASE = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
end
return 0
"""
release_script = r.register_script(RELEASE)


def acquire(key, ttl_ms):
    token = uuid.uuid4().hex
    if r.set(key, token, nx=True, px=ttl_ms):
        return token
    return None


def release_naive(key):
    r.delete(key)


def release_safe(key, token):
    return release_script(keys=[key], args=[token]) == 1


print("== 错误的释放：直接 DEL")
a = acquire("lock:build", 500)
print("A 拿到锁:", a is not None)
time.sleep(0.6)
b = acquire("lock:build", 5000)
print("A 的锁过期了，B 拿到锁:", b is not None)
release_naive("lock:build")
print("A 干完活直接 DEL，B 的锁还在吗:", r.exists("lock:build") == 1)
c = acquire("lock:build", 5000)
print("C 也拿到了锁，B 和 C 同时在跑:", c is not None)

r.flushdb()
print("== 正确的释放：先比对 token 再删")
a = acquire("lock:build", 500)
time.sleep(0.6)
b = acquire("lock:build", 5000)
print("A 用自己的 token 释放，删掉了吗:", release_safe("lock:build", a))
print("B 的锁还在吗:", r.exists("lock:build") == 1)
print("C 能拿到锁吗:", acquire("lock:build", 5000) is not None)
print("B 用自己的 token 释放:", release_safe("lock:build", b))

RENEW = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('pexpire', KEYS[1], ARGV[2])
end
return 0
"""
renew_script = r.register_script(RENEW)


def keep_alive(key, token, ttl_ms, stop):
    while not stop.wait(ttl_ms / 3000):
        if renew_script(keys=[key], args=[token, ttl_ms]) != 1:
            return


r.flushdb()
print("== 任务比锁的过期时间长：看门狗续期")
token = acquire("lock:build", 500)
stop = threading.Event()
threading.Thread(target=keep_alive, args=("lock:build", token, 500, stop), daemon=True).start()
time.sleep(1.5)
print("任务跑了 1.5 秒，锁还在吗:", r.get("lock:build") == token)
stop.set()
release_safe("lock:build", token)
print("任务结束释放后:", r.exists("lock:build") == 1)

实际输出：

== 错误的释放：直接 DEL
A 拿到锁: True
A 的锁过期了，B 拿到锁: True
A 干完活直接 DEL，B 的锁还在吗: False
C 也拿到了锁，B 和 C 同时在跑: True
== 正确的释放：先比对 token 再删
A 用自己的 token 释放，删掉了吗: False
B 的锁还在吗: True
C 能拿到锁吗: False
B 用自己的 token 释放: True
== 任务比锁的过期时间长：看门狗续期
任务跑了 1.5 秒，锁还在吗: True
任务结束释放后: False

逐段解释：

• Python 语法：三引号字符串。"""...""" 可以写多行字符串，这里用来放 Lua 脚本
• Lua 脚本：Redis 内置了 Lua 解释器，一段脚本在 Redis 里执行时不会被别的命令打断，所以”比对 + 删除”是原子的。脚本里 KEYS[1] 是传进来的第一个键名，ARGV[1] 是第一个普通参数（Lua 的下标从 1 开始）；redis.call('get', ...) 在脚本里执行一条 Redis 命令；== 是比较，if ... then ... end 是条件判断
• r.register_script(RELEASE) 把脚本注册好，返回一个可以直接调用的对象；调用时 keys=[...] 对应 KEYS，args=[...] 对应 ARGV。它会用 EVALSHA 按脚本的哈希调用，不用每次把整段脚本发过去
• uuid.uuid4().hex：生成一个随机的唯一编号，比如 3f2a...，32 个十六进制字符
• r.set(key, token, nx=True, px=ttl_ms)：对应 SET key token NX PX ttl。成功返回 True，键已存在返回 None
• Python 语法：is not None。判断一个值是不是 None，用 is，不用 ==
• 第一段：A 的锁 500 毫秒过期，睡 600 毫秒后 B 拿到锁；A 直接 DEL 把 B 的锁删了，C 又拿到锁，此时 B 和 C 都以为自己独占
• 第二段：A 用自己的旧 token 释放，脚本发现锁的值已经是 B 的，返回 0 不删；C 拿不到锁，正确
• 看门狗：keep_alive 在后台线程里每隔过期时间的三分之一（500 毫秒的三分之一约 167 毫秒）续一次期。续期脚本也先比对值，锁已经不是自己的就停止续期
• Python 语法：threading.Event。一个线程间的”信号灯”。stop.wait(秒数) 最多等这么久，期间如果别的线程调用了 stop.set() 就立刻返回 True，否则超时返回 False。所以 while not stop.wait(...) 的意思是”每隔一段时间醒一次，直到收到停止信号”
• daemon=True：守护线程，主程序退出时不用等它

几个面试会追问的细节：

什么时候用 Redis 锁就够了：锁只是为了避免重复干活、提高效率，偶尔失效只是浪费一点资源，用单个 Redis 的锁完全够。我项目防止两个构建同时改环境变量就属于这种。如果锁失效会造成数据错误，比如扣款，就要在数据库那一侧用唯一约束、乐观锁版本号兜底（锁与并发控制）。

4.5 限流与 token 配额

为什么 AI 应用特别需要限流：每个请求都在花钱，一个 Agent 任务可能花几十万 token；模型服务商本身也限流，你的服务要在自己这一层先挡住，不能把上游的配额打满。

先说这些限流算法是从哪来的。 它们不是 Web 后端发明的，而是来自通信网络。80 年代电信行业想在同一张网上同时传数据、语音、视频，碰到的问题是：用户申请了”平均每秒多少流量”，实际发起来却是一阵一阵的，一阵猛发就会把交换机的缓冲挤爆，连累别的用户。1986 年 Jonathan S. Turner 在 IEEE Communications Magazine 的文章 New Directions in Communications 里描述了后来被叫作漏桶的办法：每个连接一个计数器，发一个包加一，同时按固定速度往下减，超过阈值的包就丢掉；减的速度决定平均带宽，阈值决定能容忍多大的突发。后来 ATM 网络把它标准化成了 GCRA 算法（漏桶的来历）。令牌桶是同一个思路的另一种写法，按固定速度往桶里放令牌，拿到令牌才能发，允许把攒下的令牌一次用掉，所以能放过短时间的突发。Web 服务后来面对的是一样的问题：平均速度要限，偶尔的突发又不想一刀切，于是直接借用了这两种算法，再加上实现更简单的固定窗口和滑动窗口计数。

四种算法（口述见 3.5）：

固定窗口的边界问题画在时间轴上。限额每 10 秒 5 次，第 9.5 秒来了 5 个请求，第 10.5 秒又来 5 个：

gantt
    title 限额 10 秒 5 次，第 9.5 秒和第 10.5 秒各来 5 个请求
    dateFormat X
    axisFormat %S 秒
    section 固定窗口
    窗口一 计数 5 放行 :0, 10
    窗口二 计数 5 放行 :10, 20
    section 滑动窗口
    回看 10 秒已有 5 次 拒绝 :1, 11
    section 请求
    5 个 :milestone, 9, 0
    又 5 个 :milestone, 11, 0

（甘特图的刻度是整秒，请求的 9.5 秒和 10.5 秒画在 9 和 11 附近。）固定窗口把两批请求分到了两个窗口，各自都没超；滑动窗口在第 10.5 秒往回看 10 秒，前一批还在窗口里，所以第二批全部被拒。

动手实验 4：对比固定窗口和滑动窗口在窗口边界的表现，再做一个按 token 的每日配额。保存为 ratelimit.py。

import time
import uuid

import redis

r = redis.Redis(port=6390, decode_responses=True)
r.flushdb()


def fixed_window(user, limit, window_s, now):
    key = f"rl:fixed:{user}:{int(now // window_s)}"
    pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window_s)
    count, _ = pipe.execute()
    return count <= limit


SLIDING = """
local key, now, window, limit = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3])
redis.call('zremrangebyscore', key, '-inf', now - window)
if redis.call('zcard', key) >= limit then
    return 0
end
redis.call('zadd', key, now, ARGV[4])
redis.call('pexpire', key, window)
return 1
"""
sliding_script = r.register_script(SLIDING)


def sliding_window(user, limit, window_s, now):
    ok = sliding_script(
        keys=[f"rl:sliding:{user}"],
        args=[int(now * 1000), window_s * 1000, limit, uuid.uuid4().hex],
    )
    return ok == 1


def burst_at_boundary(fn):
    base = (int(time.time()) // 10 + 1) * 10
    passed = 0
    for t in [base - 0.5] * 5 + [base + 0.5] * 5:
        passed += fn("u1", 5, 10, t)
    return passed


print("限额：每 10 秒 5 次。在窗口边界前后 1 秒内各发 5 次，共 10 次")
print("固定窗口放行:", burst_at_boundary(fixed_window), "次")
print("滑动窗口放行:", burst_at_boundary(sliding_window), "次")

TOKEN_QUOTA = """
local used = redis.call('incrby', KEYS[1], ARGV[1])
if used == tonumber(ARGV[1]) then
    redis.call('expire', KEYS[1], ARGV[3])
end
if used > tonumber(ARGV[2]) then
    redis.call('decrby', KEYS[1], ARGV[1])
    return -1
end
return used
"""
quota_script = r.register_script(TOKEN_QUOTA)
print("每天 10 万 token 配额:")
for cost in [40000, 50000, 30000, 8000]:
    used = quota_script(keys=["quota:u1:2026-09-19"], args=[cost, 100000, 86400])
    print(f"  本次 {cost} token ->", "超额，返回 429" if used == -1 else f"已用 {used}")

实际输出：

限额：每 10 秒 5 次。在窗口边界前后 1 秒内各发 5 次，共 10 次
固定窗口放行: 10 次
滑动窗口放行: 5 次
每天 10 万 token 配额:
  本次 40000 token -> 已用 40000
  本次 50000 token -> 已用 90000
  本次 30000 token -> 超额，返回 429
  本次 8000 token -> 已用 98000

逐段解释：

• 为了能精确控制”请求发生在什么时候”，函数都接收一个 now 参数，而不是自己读当前时间。实际代码里传 time.time()
• 固定窗口的键名里带着窗口编号 int(now // window_s)：// 是整除，10 秒一个窗口，第 1789818330 秒和第 1789818339 秒属于同一个窗口
• pipeline：把多条命令攒起来一次发给 Redis，减少网络往返。pipe.execute() 返回每条命令的结果组成的列表。Python 语法：解包 count, _ = [3, True] 把列表的两个元素分别赋给两个变量，_ 表示不关心
• burst_at_boundary 算出下一个窗口边界 base，在边界前 0.5 秒发 5 次、后 0.5 秒发 5 次。Python 语法： [x] * 5 得到有 5 个 x 的列表，两个列表 + 是拼接。passed += fn(...) 里函数返回的是 True 或 False，在 Python 里可以当 1 和 0 相加
• 固定窗口：边界前后属于两个窗口，各自都没超 5 次，10 次全放行，一秒内放过了两倍
• 滑动窗口的 Lua 脚本：先用 ZREMRANGEBYSCORE 删掉一个窗口以前的记录，再用 ZCARD 数剩下几条，没超就用 ZADD 记下这次请求。分数是毫秒时间戳，成员是一个随机编号（同一毫秒的多次请求也不会互相覆盖）。整段在 Lua 里执行，“数数”和”记录”之间不会被别的请求插进来
• Lua 的 local a, b = x, y 同时定义多个局部变量；tonumber 把字符串参数转成数字，因为传进来的参数都是字符串
• token 配额脚本：INCRBY 先加上本次用量；如果是今天第一次写入（加完正好等于本次用量），就设 24 小时过期；加完超过上限就减回去并返回 -1。键名里带日期，第二天自动换新键
• 第三次 30000 被拒后，第四次 8000 仍然能过（9 万加 8 千没超），说明被拒的那次没有占用额度

实际使用时要补的：

• 返回 429 时带上 Retry-After 响应头，告诉客户端多少秒后再试
• token 用量在调用前并不知道。常见做法：调用前检查剩余额度是否够一个预估值（比如按输入长度加最大输出 token 估算），调用完按响应里的 usage 实际扣减
• 限流维度通常有好几层：按 IP 防刷、按用户防滥用、按租户做套餐配额、全局保护上游
• 网关（Nginx、云厂商的 API 网关）也能做按 IP 的限流，业务层只做和用户、token 有关的

4.6 任务队列：Redis Streams

为什么需要任务队列：口述见 3.6，14 篇 4.6 节讲过整体设计。这一节实际跑一遍。

List 做队列为什么会丢：BRPOP 取出的那一刻，任务就从 Redis 里删了。工作进程这时候崩了，任务就没了。Redis 有个”可靠队列”的老办法：用 BLMOVE 把任务从待处理列表原子地移到”处理中”列表，处理完再删掉；但超时认领要自己写。Redis 5 起有了 Streams，把这些都做好了。

先说为什么会有 Streams。 在它之前，用 Redis 存”一串按时间先后的消息”只有三个选择，Redis 作者 antirez 在 介绍 Streams 的文章里逐个说了它们的毛病：列表（List）取出就删、没有固定编号，也没法让几个消费者分着取；有序集合（Sorted Set）费内存，而且客户端不能阻塞等新消息；发布订阅（Pub/Sub）发完就忘，不在线的订阅者收不到，也查不了历史。另一边，LinkedIn 开源的 Kafka 已经把”只追加的日志 + 消费组”这套做法推广开了：消息按顺序追加、带编号永久保留，一组消费者分着读，每个消费者记住自己读到哪。antirez 读了 Kafka 的设计，把消费组的概念搬过来，按 Redis 内存数据库的特点重新做了一遍，随 2018 年 10 月的 Redis 5.0 发布。所以 Streams 可以粗略理解成”放在 Redis 里的小号 Kafka”。

Streams 的几个概念：

实验 5 的过程：

sequenceDiagram
    participant S as 任务流
    participant W1 as worker-1
    participant PEL as 待确认列表
    participant W2 as worker-2
    Note over S: 提交 zlib、libpng、curl
    S-->>W1: 读走 zlib
    S->>PEL: 记下 zlib 在 worker-1
    Note over W1: 被杀，没 XACK
    S-->>W2: 读走 libpng、curl
    W2->>PEL: XACK 这两条
    Note over PEL: 只剩 zlib
    W2->>PEL: XAUTOCLAIM
    PEL-->>W2: zlib 转给 worker-2
    W2->>PEL: 构建后 XACK
    Note over PEL: 待确认数 0

动手实验 5：三个构建任务，一个工作进程取走任务后崩溃，另一个把它认领回来；再演示重复投递时的幂等检查，以及用流做 SSE 断线续传。保存为 queue_streams.py。

import time

import redis

r = redis.Redis(port=6390, decode_responses=True)
r.flushdb()

STREAM, GROUP = "tasks:build", "workers"
r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)

for name in ["zlib", "libpng", "curl"]:
    task_id = r.xadd(STREAM, {"project": name, "target": "aarch64-linux-musl"})
    print("提交任务", name, "编号", task_id)


def handle(worker, msg_id, fields):
    if r.exists(f"done:{msg_id}"):
        print(f"  {worker}: {msg_id} 已经处理过，跳过")
    else:
        print(f"  {worker}: 构建 {fields['project']}")
        r.set(f"done:{msg_id}", worker, ex=86400)
    r.xack(STREAM, GROUP, msg_id)


print("== worker-1 取走一个任务后崩溃，没有确认")
[[_, msgs]] = r.xreadgroup(GROUP, "worker-1", {STREAM: ">"}, count=1)
crashed_id = msgs[0][0]
print("  worker-1 取到", msgs[0][1]["project"], "然后进程被杀")

print("== worker-2 正常取任务")
[[_, msgs]] = r.xreadgroup(GROUP, "worker-2", {STREAM: ">"}, count=10)
for msg_id, fields in msgs:
    handle("worker-2", msg_id, fields)

pending = r.xpending(STREAM, GROUP)
print("待确认的任务数:", pending["pending"], "在谁手里:", pending["consumers"])

time.sleep(1.1)
print("== worker-2 认领超过 1 秒没确认的任务")
_, claimed, _ = r.xautoclaim(STREAM, GROUP, "worker-2", min_idle_time=1000, start_id="0")
for msg_id, fields in claimed:
    handle("worker-2", msg_id, fields)
print("待确认的任务数:", r.xpending(STREAM, GROUP)["pending"])

print("== 同一条消息被投递两次时，幂等检查挡住重复执行")
handle("worker-3", crashed_id, {"project": "zlib"})

print("== 事件流：浏览器断线后从上次的编号续传")
ids = [r.xadd("events:run:217", {"step": i, "tool": t}) for i, t in enumerate(["clone", "cmake", "make", "check_elf"], 1)]
last_seen = ids[1]
print("浏览器最后收到的事件编号", last_seen)
for msg_id, fields in r.xread({"events:run:217": last_seen})[0][1]:
    print("  补发", msg_id, fields)

实际输出（消息编号是时间戳，每次运行不同）：

提交任务 zlib 编号 1789818334267-0
提交任务 libpng 编号 1789818334267-1
提交任务 curl 编号 1789818334267-2
== worker-1 取走一个任务后崩溃，没有确认
  worker-1 取到 zlib 然后进程被杀
== worker-2 正常取任务
  worker-2: 构建 libpng
  worker-2: 构建 curl
待确认的任务数: 1 在谁手里: [{'name': 'worker-1', 'pending': 1}]
== worker-2 认领超过 1 秒没确认的任务
  worker-2: 构建 zlib
待确认的任务数: 0
== 同一条消息被投递两次时，幂等检查挡住重复执行
  worker-3: 1789818334267-0 已经处理过，跳过
== 事件流：浏览器断线后从上次的编号续传
浏览器最后收到的事件编号 1789818335376-0
  补发 1789818335377-0 {'step': '3', 'tool': 'make'}
  补发 1789818335377-1 {'step': '4', 'tool': 'check_elf'}

逐段解释：

• xgroup_create(STREAM, GROUP, id="0", mkstream=True)：创建消费组，id="0" 表示从流的开头开始消费，mkstream=True 表示流不存在就顺便创建
• r.xadd(STREAM, {...})：追加一条消息，字段是一个字典，返回自动生成的编号
• xreadgroup(GROUP, "worker-1", {STREAM: ">"}, count=1)：以消费者 worker-1 的身份读最多 1 条。">" 表示”还没分给任何人的新消息”
• 返回值的结构是 [[流名, [(编号, 字段), ...]]]。Python 语法：嵌套解包 [[_, msgs]] = ... 按同样的形状拆开，_ 接住流名，msgs 接住消息列表。msgs[0][0] 是第一条消息的编号，msgs[0][1] 是它的字段
• Python 语法：for msg_id, fields in msgs: 每次循环把一个 (编号, 字段) 拆成两个变量
• worker-1 读走 zlib 后什么都没做，模拟进程被杀。xpending 显示有 1 条待确认，在 worker-1 手里
• xautoclaim(..., min_idle_time=1000, start_id="0")：把空闲超过 1000 毫秒的待确认消息认领给 worker-2。返回三个值：下次扫描的起点、认领到的消息、已被删除的消息编号，用 _, claimed, _ 只取中间那个
• handle 的幂等检查：处理前查 done:编号 是否存在，处理成功之后再写这个标记。worker-3 又收到 zlib 那条消息时，发现已经完成，只确认不重做
• 这里有个我自己写错过的地方：一开始我把标记写在处理之前（用 SET NX 抢占），结果如果处理到一半崩了，重试时会以为已经做完，任务就丢了。标记必须在处理成功后写。真正严格的做法还要处理”两个消费者同时处理同一条”的情况，可以在处理期间再加一把按任务编号的锁
• 事件流部分：enumerate(列表, 1) 同时给出序号（从 1 开始）和元素。Python 语法： 列表推导式里 for i, t in enumerate(...) 一次拿两个值。xread({"events:run:217": last_seen}) 读编号大于 last_seen 的所有消息，这正好是 SSE 断线重连时要做的：浏览器重连带上 Last-Event-ID，服务端从那之后补发

真实项目里用什么：

别用发布订阅做任务队列：Redis 的 PUBLISH / SUBSCRIBE 不存消息，发的时候订阅者不在线就收不到。它适合”通知在线的进程”，比如工作进程产生事件、通知所有 Web 进程推给各自的浏览器。

4.7 鉴权：登录、JWT、密码、API key、多租户

先分清两件事：

• 认证（Authentication）：确认你是谁。登录、验 token、验 API key 都是认证。失败返回 401
• 授权（Authorization）：确认你能做什么。管理员才能建 API key、只能看自己租户的数据，都是授权。失败返回 403

先说为什么会有 Session，后来又有了 JWT。 HTTP 本身不记人：第二个请求来的时候，服务器不知道它和第一个请求是同一个人发的。1994 年 Netscape 的 Lou Montulli 为了给网上商店做购物车，发明了 Cookie：服务器在响应里让浏览器存一小段数据，之后浏览器每次请求都自动带回来（HTTP cookie 的历史）。登录也就顺理成章地变成了：登录成功后服务器生成一个随机的会话编号，放进 Cookie，服务器自己在内存或数据库里记下”这个编号是谁”，这就是 Session。

Session 的麻烦出在服务变多以后。服务器有好几台，会话记录要放到大家都能查的 Redis 里，每个请求都得查一次；一个登录要给好几个不同公司、不同域名的服务用，Cookie 又跨不了域。于是有人想：能不能把”这是谁、有效到几点”直接写在令牌里，签上名，谁拿到都能自己验签，不用回头查存储？这就是 JWT。它出自 IETF 的 OAuth 工作组，2015 年 5 月发布为 RFC 7519，作者是微软的 Michael B. Jones、Ping Identity 的 John Bradley 和野村综合研究所的 Nat Sakimura。代价也是从这里来的：令牌发出去就收不回来，要强制下线还得再加一层作废名单（本节最后的表）。

Session 和 JWT：口述见 3.7。

JWT 的结构（RFC 7519）：头部.载荷.签名 三段，每段都是 Base64URL 编码。

登录和之后每次请求的流程：

sequenceDiagram
    participant B as 客户端
    participant API as FastAPI
    participant DB as 用户表
    B->>API: POST /login 用户名 + 密码
    API->>DB: 取出密码哈希
    API->>API: Argon2 验证密码
    API-->>B: JWT（sub、role、tenant、exp，用密钥签名）
    B->>API: GET /runs，请求头 Authorization: Bearer JWT
    API->>API: 验签、检查过期（不查任何存储）
    API->>DB: 查询时带上令牌里的 tenant
    API-->>B: 只返回本租户的数据

先说密码哈希为什么一换再换。 最早的系统直接存明文，数据库一泄露密码全丢。后来改存 MD5、SHA 这类哈希，可这类函数设计目标就是算得快，攻击者拿到哈希后用显卡每秒能试上亿个候选密码。1999 年 OpenBSD 项目的 Niels Provos 和 David Mazières 在 USENIX 会议上发表了 bcrypt，核心想法是故意算得慢，而且慢的程度可以调：硬件变快了就把成本参数调大，破解一次的代价跟着涨。但 bcrypt 只慢在计算上，攻击者用显卡、专用芯片并行跑照样能提速。2013 到 2015 年密码学界办了一次公开的密码哈希竞赛，24 个候选算法里选出了卢森堡大学 Alex Biryukov、Daniel Dinu、Dmitry Khovratovich 设计的 Argon2（2015 年 7 月公布，竞赛官网），它除了慢还要占大量内存，内存很难像算力那样便宜地堆上去；2021 年它又整理成了 RFC 9106。下面实验用的 Argon2id 就是它的一个变体。

密码和 API key 怎么存：口述见 3.8。

动手实验 6：一个带登录的 FastAPI 小服务：用户名密码登录拿 JWT，按角色控制权限，按租户过滤数据，管理员能创建 API key。用 FastAPI 自带的测试客户端直接调，不用启动服务器。依赖：pip install fastapi pyjwt "pwdlib[argon2]" python-multipart httpx。保存为 auth_app.py。

import base64
import hashlib
import json
import secrets
from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.testclient import TestClient
from pwdlib import PasswordHash

SECRET = secrets.token_hex(32)
ALGORITHM = "HS256"
hasher = PasswordHash.recommended()

USERS = {
    "alice": {"hash": hasher.hash("alice-pass"), "role": "admin", "tenant": "t1"},
    "bob": {"hash": hasher.hash("bob-pass"), "role": "member", "tenant": "t2"},
}
RUNS = [
    {"id": 1, "tenant": "t1", "project": "zlib"},
    {"id": 2, "tenant": "t2", "project": "curl"},
    {"id": 3, "tenant": "t1", "project": "libpng"},
]
API_KEYS = {}

app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="/login")


def make_token(username, minutes=30):
    user = USERS[username]
    payload = {
        "sub": username,
        "role": user["role"],
        "tenant": user["tenant"],
        "exp": datetime.now(timezone.utc) + timedelta(minutes=minutes),
    }
    return jwt.encode(payload, SECRET, algorithm=ALGORITHM)


def current_user(token: str = Depends(oauth2)):
    try:
        payload = jwt.decode(token, SECRET, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "登录已过期")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "token 无效")
    return payload


def require_role(role):
    def checker(user=Depends(current_user)):
        if user["role"] != role:
            raise HTTPException(403, "没有权限")
        return user
    return checker


@app.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = USERS.get(form.username)
    if not user or not hasher.verify(form.password, user["hash"]):
        raise HTTPException(401, "用户名或密码错误")
    return {"access_token": make_token(form.username), "token_type": "bearer"}


@app.get("/runs")
def list_runs(user=Depends(current_user)):
    return [run for run in RUNS if run["tenant"] == user["tenant"]]


@app.post("/admin/api-keys")
def create_api_key(user=Depends(require_role("admin"))):
    raw = "sk-" + secrets.token_urlsafe(24)
    API_KEYS[hashlib.sha256(raw.encode()).hexdigest()] = user["tenant"]
    return {"api_key": raw, "note": "只显示这一次"}


@app.get("/v1/runs")
def list_runs_by_key(x_api_key: str = Header()):
    tenant = API_KEYS.get(hashlib.sha256(x_api_key.encode()).hexdigest())
    if tenant is None:
        raise HTTPException(401, "API key 无效")
    return [run for run in RUNS if run["tenant"] == tenant]


client = TestClient(app)

print("同一个密码哈希两次，结果不同（每次随机加盐）:")
print(" ", hasher.hash("alice-pass")[:60], "...")
print(" ", hasher.hash("alice-pass")[:60], "...")

print("错误密码登录:", client.post("/login", data={"username": "alice", "password": "x"}).status_code)
token = client.post("/login", data={"username": "alice", "password": "alice-pass"}).json()["access_token"]
payload_part = token.split(".")[1]
print("JWT 中间段不用密钥就能解出来:", json.loads(base64.urlsafe_b64decode(payload_part + "==")))

auth = {"Authorization": f"Bearer {token}"}
print("不带 token 查运行记录:", client.get("/runs").status_code)
print("alice 查运行记录（只看到 t1 的）:", client.get("/runs", headers=auth).json())

bob = client.post("/login", data={"username": "bob", "password": "bob-pass"}).json()["access_token"]
print("bob 创建 API key:", client.post("/admin/api-keys", headers={"Authorization": f"Bearer {bob}"}).status_code)
key = client.post("/admin/api-keys", headers=auth).json()["api_key"]
print("alice 创建 API key 成功，服务端只存哈希:", list(API_KEYS)[0][:16], "...")
print("用 API key 调接口:", client.get("/v1/runs", headers={"X-API-Key": key}).json())

tampered = token[:-4] + ("AAAA" if not token.endswith("AAAA") else "BBBB")
print("改过签名的 token:", client.get("/runs", headers={"Authorization": f"Bearer {tampered}"}).json())
expired = jwt.encode({"sub": "alice", "role": "admin", "tenant": "t1",
                      "exp": datetime.now(timezone.utc) - timedelta(seconds=1)}, SECRET, algorithm=ALGORITHM)
print("过期的 token:", client.get("/runs", headers={"Authorization": f"Bearer {expired}"}).json())

实际输出（fastapi 0.141.1，PyJWT 2.14.0，pwdlib 用 Argon2；哈希和时间戳每次不同）：

同一个密码哈希两次，结果不同（每次随机加盐）:
  $argon2id$v=19$m=65536,t=3,p=4$Pm6lwJejqhFePQmy5NO5pg$4uYjuV ...
  $argon2id$v=19$m=65536,t=3,p=4$/WYlEmfGrXYJIzs1IENjIQ$cvb8dx ...
错误密码登录: 401
JWT 中间段不用密钥就能解出来: {'sub': 'alice', 'role': 'admin', 'tenant': 't1', 'exp': 1789820672}
不带 token 查运行记录: 401
alice 查运行记录（只看到 t1 的）: [{'id': 1, 'tenant': 't1', 'project': 'zlib'}, {'id': 3, 'tenant': 't1', 'project': 'libpng'}]
bob 创建 API key: 403
alice 创建 API key 成功，服务端只存哈希: abd6b6fdf82c3b0b ...
用 API key 调接口: [{'id': 1, 'tenant': 't1', 'project': 'zlib'}, {'id': 3, 'tenant': 't1', 'project': 'libpng'}]
改过签名的 token: {'detail': 'token 无效'}
过期的 token: {'detail': '登录已过期'}

逐段解释：

• secrets.token_hex(32)：生成 32 字节的随机密钥，用十六进制表示。secrets 模块专门用来生成安全的随机数，不要用 random 生成密钥和 token，后者可以被预测。真实项目里密钥从环境变量读，不能每次启动随机生成，否则一重启所有人都被踢下线
• PasswordHash.recommended()：pwdlib 推荐的配置，也就是 Argon2id。FastAPI 官方教程现在推荐 PyJWT 加 pwdlib（文档）
• 哈希结果 $argon2id$v=19$m=65536,t=3,p=4$盐$哈希值：依次是算法、版本、内存 64 MiB、迭代 3 次、并行度 4、随机盐、结果。参数和盐都存在这个字符串里，验证时 hasher.verify(明文, 哈希) 会自己解析。OWASP 的最低要求是 19 MiB、2 次迭代、并行度 1（密码存储速查表），这个默认值高于最低要求
• 同一个密码两次哈希结果不同，是因为每次的盐不同。这样攻击者不能用预先算好的”密码 → 哈希”对照表（彩虹表）
• Python 语法：from ... import ...。from datetime import datetime, timedelta, timezone 从 datetime 模块里导入三个名字，之后直接用 datetime.now(...)，不用写模块名
• datetime.now(timezone.utc) + timedelta(minutes=30)：当前 UTC 时间加 30 分钟。exp 字段 PyJWT 会自动转成时间戳，解码时自动检查是否过期
• Depends（依赖注入）：FastAPI 的核心机制。def list_runs(user=Depends(current_user)) 的意思是：处理这个请求之前，先调用 current_user，把它的返回值作为 user 参数传进来；如果 current_user 抛了异常，接口函数根本不会执行。鉴权逻辑写一次，所有接口都能复用
• OAuth2PasswordBearer(tokenUrl="/login")：它本身也是一个依赖，作用是从请求头 Authorization: Bearer xxx 里取出 xxx；没有这个头就直接返回 401。tokenUrl 只是告诉自动生成的接口文档去哪里登录
• jwt.decode(token, SECRET, algorithms=[ALGORITHM])：验签并解码。algorithms 一定要显式指定，这是防止攻击者在头部把算法改成 none 之类来绕过验签
• Python 语法：多个 except。try 里出错后，按顺序匹配 except 后面的异常类型。ExpiredSignatureError 是 InvalidTokenError 的子类，所以要写在前面，否则会被后者先接住，过期也提示”token 无效”
• Python 语法：raise。主动抛出异常。FastAPI 捕获 HTTPException 后，把状态码和 detail 变成响应返回
• 依赖一层套一层，请求进来时按这个顺序执行，任何一层抛异常，后面的都不会执行：

flowchart TB
    R[请求] --> O["oauth2<br/>取出 Bearer token<br/>没有就 401"]
    O --> U["current_user<br/>验签、查过期<br/>失败就 401"]
    U --> C["require_role('admin')<br/>角色不对就 403"]
    C --> E["create_api_key<br/>真正的接口逻辑"]

• require_role(role) 是一个返回函数的函数。Python 语法：闭包。里面定义的 checker 记住了外层的 role，所以 require_role("admin") 得到一个”检查是不是管理员”的依赖。它自己又依赖 current_user，依赖可以一层套一层：先认证，再授权
• OAuth2PasswordRequestForm = Depends()：从表单里取 username 和 password。OAuth2 规范要求登录用表单提交，所以需要 python-multipart 包
• hasher.verify(...) 之前先判断 not user：用户不存在和密码错误返回同样的提示，不告诉对方”这个用户名存在”
• Python 语法：列表推导式带条件。[run for run in RUNS if run["tenant"] == user["tenant"]] 只保留租户匹配的记录。租户编号来自令牌，不来自请求参数，这是多租户隔离的关键：如果写成 GET /runs?tenant=t2，用户改一下参数就能看别人的数据
• Header()：从请求头取值。参数名 x_api_key 会自动对应请求头 X-API-Key（下划线换成横线，大小写不敏感）
• API key：secrets.token_urlsafe(24) 生成随机串，加上 sk- 前缀，只在创建时返回一次，服务端只存 SHA-256 哈希。请求来了把收到的 key 算一次哈希去字典里查。数据库泄露时攻击者拿到的只是哈希，用不了
• JWT 中间段：token.split(".")[1] 取出载荷段，补上 Base64 的填充 == 后直接解码，不需要任何密钥。这证明载荷只是编码、没有加密
• 改了签名最后几个字符，验签失败，401；手工造一个 1 秒前就过期的 token，提示”登录已过期”

实际项目要补的：

4.8 MySQL：表设计、EXPLAIN、慢查询

索引原理、B+ 树、事务和锁在 cs-fundamentals/03-数据库 讲过，这一节用实际数据验证那些结论，再补表设计和慢查询排查。

先说为什么后端大多用 MySQL。 90 年代中期要给网站配数据库，能选的主要是 Oracle、DB2 这类商业数据库，要付授权费，部署也重。瑞典的 David Axmark、Allan Larsson 和芬兰的 Michael “Monty” Widenius 从 1994 年开始开发 MySQL，1995 年 5 月 23 日发布第一个版本（MySQL 的历史），名字里的 My 是 Widenius 女儿的名字。它免费、开源、装起来简单，和 Linux、Apache、PHP 一起组成了当年做网站的标准搭配 LAMP，很多互联网公司就是这么起步的，后端工程师的经验和工具也就大量积累在它上面。它后来几经转手，2008 年 Sun 收购了 MySQL AB，2010 年 Oracle 收购 Sun 后成了 Oracle 的产品，Widenius 在 2009 年另外分出了 MariaDB。现在 MySQL 默认的存储引擎 InnoDB 支持事务和行锁，是 2010 年 12 月正式发布的 5.5 版才改成默认的，之前默认的 MyISAM 不支持事务。本节后面讲的索引、EXPLAIN 都是针对 InnoDB。

一个 AI 应用的典型表（CrossBuild 的 SQLite 里有 runs 和 tool_calls 两张，下面按多用户的版本扩展）：

几个设计上的考虑：多租户的表都带 tenant_id，而且索引以它开头；工具参数用 JSON 类型存，结构不固定；工具结果和模型原始响应可能很大，不要放在经常查询的表里，另开一张表或放对象存储；时间字段统一用 UTC。

动手实验 7：造 2 万条运行记录和 30 万条工具调用，对比加索引前后。先用 SQL 造数据，保存为 mysql_setup.sql：

DROP DATABASE IF EXISTS agentlab;
CREATE DATABASE agentlab DEFAULT CHARSET utf8mb4;
USE agentlab;

CREATE TABLE runs (
    id          BIGINT PRIMARY KEY AUTO_INCREMENT,
    tenant_id   INT NOT NULL,
    project     VARCHAR(64) NOT NULL,
    status      VARCHAR(16) NOT NULL,
    started_at  DATETIME NOT NULL,
    steps       INT NOT NULL DEFAULT 0
);

CREATE TABLE tool_calls (
    id           BIGINT PRIMARY KEY AUTO_INCREMENT,
    run_id       BIGINT NOT NULL,
    step         INT NOT NULL,
    tool         VARCHAR(32) NOT NULL,
    ok           TINYINT NOT NULL,
    duration_ms  INT NOT NULL,
    result       TEXT
);

SET SESSION cte_max_recursion_depth = 1000000;

INSERT INTO runs (tenant_id, project, status, started_at, steps)
WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 20000)
SELECT n % 50,
       ELT(1 + n % 5, 'zlib', 'libpng', 'curl', 'json', 'libarchive'),
       ELT(1 + CRC32(n) % 10, 'failed', 'passed', 'passed', 'passed', 'passed', 'passed', 'passed', 'passed', 'passed', 'error'),
       TIMESTAMP('2026-01-01') + INTERVAL n * 13 MINUTE,
       15
FROM seq;

INSERT INTO tool_calls (run_id, step, tool, ok, duration_ms, result)
WITH RECURSIVE seq(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM seq WHERE n < 299999)
SELECT 1 + n DIV 15,
       1 + n % 15,
       ELT(1 + n % 6, 'read_file', 'run_command', 'write_file', 'list_dir', 'search_knowledge', 'check_elf'),
       n % 7 <> 0,
       10 + n % 5000,
       REPEAT('x', 200)
FROM seq;

ANALYZE TABLE runs, tool_calls;
SELECT (SELECT COUNT(*) FROM runs) AS runs, (SELECT COUNT(*) FROM tool_calls) AS tool_calls;

这段 SQL 里的新写法：

• WITH RECURSIVE seq(n) AS (...)：递归公用表表达式，这里用来生成 1 到 20000 的数字序列。SELECT 1 是起点，SELECT n + 1 FROM seq WHERE n < 20000 不断在上一行基础上加一
• ELT(下标, 值1, 值2, ...)：按下标取值，下标从 1 开始
• CRC32(n) % 10：用校验和打散，让状态分布和租户编号无关。我第一次用 n % 10 造状态，结果它和 n % 50 的租户编号绑在一起，每个租户只有一种状态，测出来的数字不可信，后来才改成这样
• ANALYZE TABLE：更新表的统计信息，优化器靠它估计行数

然后用 Python 逐个对比，依赖 pip install pymysql。保存为 mysql_explain.py：

import time

import pymysql

conn = pymysql.connect(host="127.0.0.1", port=3307, user="root",
                       database="agentlab", cursorclass=pymysql.cursors.DictCursor)
cur = conn.cursor()


def explain(sql, params=()):
    cur.execute("EXPLAIN FORMAT=TRADITIONAL " + sql, params)
    row = cur.fetchone()
    return {k: row[k] for k in ("type", "key", "rows", "Extra")}


def timed(sql, params=(), repeat=5):
    start = time.perf_counter()
    for _ in range(repeat):
        cur.execute(sql, params)
        cur.fetchall()
    return (time.perf_counter() - start) / repeat * 1000


def show(title, sql, params=()):
    print(f"{title}\n  {explain(sql, params)}\n  平均 {timed(sql, params):.1f} ms")


def drop_index(table, name):
    cur.execute("SELECT 1 FROM information_schema.statistics WHERE table_schema='agentlab' "
                "AND table_name=%s AND index_name=%s", (table, name))
    if cur.fetchone():
        cur.execute(f"ALTER TABLE {table} DROP INDEX {name}")


drop_index("tool_calls", "idx_run")
drop_index("runs", "idx_tenant_status_time")
drop_index("runs", "idx_started")

q1 = "SELECT step, tool, ok FROM tool_calls WHERE run_id = %s ORDER BY id"
show("【1】查一次运行的全部工具调用，run_id 没有索引", q1, (12345,))
cur.execute("ALTER TABLE tool_calls ADD INDEX idx_run (run_id)")
cur.execute("ANALYZE TABLE tool_calls")
cur.fetchall()
show("【1】给 run_id 加索引后", q1, (12345,))

q2 = ("SELECT id, project, started_at FROM runs WHERE tenant_id = %s AND status = 'failed' "
      "ORDER BY started_at DESC LIMIT 20")
show("【2】租户的失败记录，按时间倒序，没有索引", q2, (7,))
cur.execute("ALTER TABLE runs ADD INDEX idx_tenant_status_time (tenant_id, status, started_at)")
cur.execute("ANALYZE TABLE runs")
cur.fetchall()
show("【2】加联合索引 (tenant_id, status, started_at) 后", q2, (7,))

q3 = "SELECT id, project FROM runs WHERE status = 'failed' ORDER BY started_at DESC LIMIT 20"
show("【3】跳过最左列 tenant_id，只按 status 查", q3)

q4 = "SELECT tenant_id, status, started_at FROM runs WHERE tenant_id = %s AND status = 'error'"
show("【4】查的列都在索引里（覆盖索引）", q4, (7,))

drop_index("runs", "idx_started")
cur.execute("ALTER TABLE runs ADD INDEX idx_started (started_at)")
cur.execute("ANALYZE TABLE runs")
cur.fetchall()
q5a = "SELECT id FROM runs WHERE DATE(started_at) = '2026-03-01'"
q5b = "SELECT id FROM runs WHERE started_at >= '2026-03-01' AND started_at < '2026-03-02'"
show("【5】对索引列 started_at 套函数 DATE()", q5a)
show("【5】改写成范围条件", q5b)

q6a = "SELECT id, tool FROM tool_calls ORDER BY id LIMIT 280000, 20"
q6b = "SELECT id, tool FROM tool_calls WHERE id > %s ORDER BY id LIMIT 20"
show("【6】深分页 LIMIT 280000, 20", q6a)
show("【6】按上一页最后一个 id 往后取", q6b, (280000,))

实际输出（MySQL 26.7，Apple M 系列本机，每条跑 5 次取平均）：

【1】查一次运行的全部工具调用，run_id 没有索引
  {'type': 'index', 'key': 'PRIMARY', 'rows': 294988, 'Extra': 'Using where'}
  平均 57.3 ms
【1】给 run_id 加索引后
  {'type': 'ref', 'key': 'idx_run', 'rows': 15, 'Extra': None}
  平均 0.2 ms
【2】租户的失败记录，按时间倒序，没有索引
  {'type': 'ALL', 'key': None, 'rows': 20306, 'Extra': 'Using where; Using filesort'}
  平均 4.2 ms
【2】加联合索引 (tenant_id, status, started_at) 后
  {'type': 'ref', 'key': 'idx_tenant_status_time', 'rows': 39, 'Extra': 'Backward index scan'}
  平均 0.2 ms
【3】跳过最左列 tenant_id，只按 status 查
  {'type': 'ALL', 'key': None, 'rows': 20306, 'Extra': 'Using where; Using filesort'}
  平均 3.9 ms
【4】查的列都在索引里（覆盖索引）
  {'type': 'ref', 'key': 'idx_tenant_status_time', 'rows': 38, 'Extra': 'Using index'}
  平均 0.3 ms
【5】对索引列 started_at 套函数 DATE()
  {'type': 'index', 'key': 'idx_started', 'rows': 20306, 'Extra': 'Using where; Using index'}
  平均 2.5 ms
【5】改写成范围条件
  {'type': 'range', 'key': 'idx_started', 'rows': 111, 'Extra': 'Using where; Using index'}
  平均 0.3 ms
【6】深分页 LIMIT 280000, 20
  {'type': 'index', 'key': 'PRIMARY', 'rows': 280020, 'Extra': None}
  平均 36.7 ms
【6】按上一页最后一个 id 往后取
  {'type': 'range', 'key': 'PRIMARY', 'rows': 39748, 'Extra': 'Using where'}
  平均 0.3 ms

代码里的新写法：

• cursorclass=pymysql.cursors.DictCursor：查询结果每行是一个字典，可以按列名取值
• %s 占位符：cur.execute(sql, (12345,)) 由驱动负责把参数安全地放进 SQL，防止 SQL 注入。不能用 f-string 把用户输入拼进 SQL。注意这里的 %s 是 pymysql 的占位符，不是 Python 的字符串格式化；表名、索引名不能用占位符，所以 drop_index 里用了 f-string，但那两个值是写死在代码里的，不来自用户
• Python 语法：字典推导式。{k: row[k] for k in ("type", "key", "rows", "Extra")} 从一行结果里挑出这四列组成新字典
• EXPLAIN FORMAT=TRADITIONAL：我装的 MySQL 26.7 默认输出树形格式，加上这个才是大多数文章里的表格格式。想看实际执行的耗时和行数，可以用 EXPLAIN ANALYZE

逐组解读：

把四组的耗时画在一起（毫秒，每条 5 次平均）：

xychart-beta
    title "改之前 vs 改之后的平均耗时（毫秒）"
    x-axis ["1 无索引", "1 加索引", "2 无索引", "2 联合索引", "5 套函数", "5 改范围", "6 偏移分页", "6 游标分页"]
    y-axis "毫秒" 0 --> 60
    bar [57.3, 0.2, 4.2, 0.2, 2.5, 0.3, 36.7, 0.3]

深分页为什么慢，看索引的叶子节点就明白了：

flowchart LR
    subgraph OFF["LIMIT 280000, 20"]
        direction TB
        L1[第 1 行] --> L2[第 2 行] --> L3["……逐行往后数<br/>28 万行全部读过再丢掉"] --> L4[返回第 280001～280020 行]
    end
    subgraph KEY["WHERE id > 280000 LIMIT 20"]
        direction TB
        ROOT[B+ 树根节点] --> MID[中间节点] --> LEAF["直接定位到 id=280001"] --> T20[往后读 20 行就停]
    end
    OFF ~~~ KEY

慢查询日志：线上不能对每条 SQL 做 EXPLAIN，先靠慢查询日志找出慢的那些。实验里把阈值设成 0.02 秒，然后分别跑深分页和游标分页各一次：

SET GLOBAL slow_query_log_file = '/path/to/slow.log';
SET GLOBAL long_query_time = 0.02;
SET GLOBAL slow_query_log = ON;

日志里只记下了深分页那条：

# Query_time: 0.036802  Lock_time: 0.000005 Rows_sent: 20  Rows_examined: 280020
SELECT id, tool FROM tool_calls ORDER BY id LIMIT 280000, 20;

Rows_examined 28 万、Rows_sent 20，扫描行数和返回行数差了一万多倍，一眼就能看出问题。SET GLOBAL 重启后失效，要长期开就写进配置文件；日志多的时候用 mysqldumpslow 或 Percona 的 pt-query-digest 按 SQL 模式汇总。

还有两个应用层的常见问题：

• N+1 查询：先查出 20 个运行，再在循环里对每个运行查一次工具调用，一共 21 次查询。用 ORM 时很容易不知不觉写出来。改成一次 WHERE run_id IN (...) 批量查，或者一次 JOIN
• 连接池：每次请求都新建数据库连接很慢（TCP 握手、认证）。连接池预先建好一批连接，用完放回去。池子的上限乘以实例数，要小于数据库的 max_connections，否则会报 Too many connections。SQLAlchemy 默认就带连接池

4.9 稳定性：熔断、降级、舱壁

超时、重试、退避、错误分类在 02 篇 讲过。这一节补三个：熔断、降级、舱壁。口述见 3.11。

先说为什么会有熔断和舱壁。 只有超时和重试时，下游一挂会发生这样的事：每个请求都要等满超时才失败，失败了还要重试几次，等待中的请求占着线程和连接越积越多，最后自己的服务也被拖垮，再往上的调用方跟着挂，一层传一层，这叫级联故障。Michael Nygard 在 2007 年出版的《Release It!》里总结了一批让线上系统”扛得住”的做法，其中就有这两个，名字都借自现实：熔断器借自家里的电闸，电流异常时自动断开，保护整条线路；舱壁借自船舱的隔板，一个舱进水只淹一个舱，船不沉。让它们在业界普及的是 Netflix：2012 年 11 月他们开源了 Java 库 Hystrix（发布文章），给每个下游依赖配一个熔断器和一个独立的线程池或信号量。Martin Fowler 2014 年的短文 CircuitBreaker 又把三种状态讲成了现在通行的样子。Hystrix 本身已经停止开发，Netflix 在 README 里建议新项目改用 resilience4j，但这套思路留了下来。

熔断器的三种状态（微软架构文档）：

stateDiagram-v2
    [*] --> 关闭
    关闭 --> 打开: 连续失败达到阈值
    打开 --> 半开: 等待一段时间后
    半开 --> 关闭: 试探请求成功
    半开 --> 打开: 试探请求失败

动手实验 8：一个最小的熔断器，加上降级到备用模型，再用信号量做舱壁。保存为 breaker.py。

import asyncio
import time


class CircuitOpen(Exception):
    pass


class CircuitBreaker:
    def __init__(self, fail_threshold=3, reset_after=1.0):
        self.fail_threshold = fail_threshold
        self.reset_after = reset_after
        self.failures = 0
        self.state = "closed"
        self.opened_at = 0.0

    async def call(self, fn):
        if self.state == "open":
            if time.monotonic() - self.opened_at < self.reset_after:
                raise CircuitOpen()
            self.state = "half_open"
        try:
            result = await fn()
        except Exception:
            self.failures += 1
            if self.state == "half_open" or self.failures >= self.fail_threshold:
                self.state = "open"
                self.opened_at = time.monotonic()
            raise
        self.failures = 0
        self.state = "closed"
        return result


upstream = {"healthy": False, "hits": 0}


async def call_primary_model():
    upstream["hits"] += 1
    await asyncio.sleep(0.05)
    if not upstream["healthy"]:
        raise TimeoutError("主模型超时")
    return "主模型的回答"


async def call_backup_model():
    return "备用模型的回答"


breaker = CircuitBreaker()


async def ask(i):
    try:
        answer = await breaker.call(call_primary_model)
    except CircuitOpen:
        answer = await call_backup_model() + "（熔断中，没打主模型）"
    except TimeoutError:
        answer = await call_backup_model() + "（主模型失败，降级）"
    print(f"  请求 {i}: 状态 {breaker.state:9} -> {answer}")


async def main():
    print("== 主模型挂了")
    for i in range(1, 7):
        await ask(i)
    print("主模型实际被调用", upstream["hits"], "次")
    await asyncio.sleep(1.1)
    upstream["healthy"] = True
    print("== 1 秒后主模型恢复，放一个试探请求")
    for i in range(7, 9):
        await ask(i)

    print("== 舱壁：最多同时 3 个请求打到模型")
    limit = asyncio.Semaphore(3)
    running = {"now": 0, "max": 0}

    async def limited_call():
        async with limit:
            running["now"] += 1
            running["max"] = max(running["max"], running["now"])
            await asyncio.sleep(0.1)
            running["now"] -= 1

    start = time.perf_counter()
    await asyncio.gather(*(limited_call() for _ in range(10)))
    print(f"  10 个请求，同时在跑的最多 {running['max']} 个，总耗时 {time.perf_counter() - start:.2f}s")


asyncio.run(main())

实际输出：

== 主模型挂了
  请求 1: 状态 closed    -> 备用模型的回答（主模型失败，降级）
  请求 2: 状态 closed    -> 备用模型的回答（主模型失败，降级）
  请求 3: 状态 open      -> 备用模型的回答（主模型失败，降级）
  请求 4: 状态 open      -> 备用模型的回答（熔断中，没打主模型）
  请求 5: 状态 open      -> 备用模型的回答（熔断中，没打主模型）
  请求 6: 状态 open      -> 备用模型的回答（熔断中，没打主模型）
主模型实际被调用 3 次
== 1 秒后主模型恢复，放一个试探请求
  请求 7: 状态 closed    -> 主模型的回答
  请求 8: 状态 closed    -> 主模型的回答
== 舱壁：最多同时 3 个请求打到模型
  10 个请求，同时在跑的最多 3 个，总耗时 0.41s

逐段解释：

• Python 语法：自定义异常。class CircuitOpen(Exception): pass 定义一个新的异常类型，继承自 Exception，pass 表示什么都不加。有了专门的类型，调用方就能用 except CircuitOpen 单独处理”熔断中”这种情况
• Python 语法：类。class CircuitBreaker: 定义一个类，把状态（失败次数、当前状态）和操作（call）放在一起。__init__ 是创建对象时自动调用的初始化方法；self 指这个对象本身，self.failures = 0 就是给对象设一个属性。breaker = CircuitBreaker() 创建一个对象
• async def call(self, fn)：fn 是一个异步函数，await fn() 调用它并等结果
• time.monotonic()：单调递增的时钟，专门用来算时间间隔，不受系统改时间影响
• 打开状态下，距离打开还不到 reset_after 秒就直接抛 CircuitOpen，根本不调用下游；超过了就变成半开，放这一个请求过去试探
• except Exception: 里最后的 raise 不带参数，表示把刚才捕获的异常原样再抛出去，让调用方知道失败了
• 输出里请求 3 是第 3 次失败，熔断器在这次失败后打开，所以打印时状态已经是 open；请求 4 到 6 没有调用主模型，主模型总共只被调了 3 次
• 1.1 秒后请求 7 进入半开状态，试探成功，恢复关闭
• Python 语法：格式说明 {breaker.state:9} 表示占 9 个字符宽，让输出对齐
• 舱壁：asyncio.Semaphore(3) 是一个有 3 个名额的信号量。Python 语法：async with limit: 进入时拿一个名额，拿不到就等；离开缩进块时自动归还。10 个请求每个 0.1 秒，同时最多 3 个，所以大约分 4 批，总耗时 0.41 秒
• Python 语法：asyncio.gather(*(...))。(limited_call() for _ in range(10)) 是生成器表达式，产生 10 个协程；前面的 * 把它们展开成 10 个独立参数传给 gather，gather 让它们并发运行并等全部完成

这个实现省略了什么：半开状态下同时来了多个请求，会全部被放过去试探，正式的实现只放一个；多实例部署时，每个实例各有一个熔断器，状态不共享，这通常可以接受。Python 里现成的库有 pybreaker、aiobreaker；很多团队在网关或服务网格那一层做熔断。

降级要提前准备好：备用模型要提前评测过（09 篇），不能等出事了才临时切一个没测过的；降级时要在日志和监控里标出来，否则”一直在用备用模型”这件事没人发现。

4.10 部署：Docker、docker-compose、Nginx

口述见 3.12。这一节用一个小服务把三件事都跑一遍：API（FastAPI）、Redis、Nginx，并在 Nginx 后面实测 SSE 的几个坑，最后验证多 worker 下进程内锁失效。

Docker 和 Nginx 各自是为了解决什么问题出现的，在 14 篇 4.7 节讲过。这里补一下 docker-compose 是怎么来的：Docker 刚出来时一次只管一个容器，一个应用要 API、Redis、数据库好几个容器一起跑，就得手敲好几条很长的 docker run，还要记住谁先启动、谁连谁、端口怎么映射。伦敦一家叫 Orchard 的小公司（Ben Firshman 和 Aanand Prasad 创办）在 2013 年底做了一个叫 Fig 的工具，把这些写进一个 YAML 文件，一条命令全部拉起来。2014 年 7 月 Docker 公司收购了 Orchard（报道），Fig 在同年 10 月发布 1.0，之后改名为 Docker Compose（Docker 的历史）。现在 Compose 已经做进了 docker 命令本身，下面的命令写成 docker compose，老写法 docker-compose 是早期单独安装的版本。

几个概念：

• 镜像（image）：一个打包好的只读文件系统，里面有操作系统基础文件、Python、依赖和你的代码
• 容器（container）：镜像跑起来的实例，有自己独立的文件系统、网络和进程空间
• Dockerfile：描述怎么一步步构建镜像。每条指令产生一层，构建时如果某一层的输入没变，就直接用缓存
• docker-compose：用一个 YAML 文件描述多个容器怎么一起跑，一条命令全部启动。同一个 compose 里的服务可以用服务名互相访问，比如 API 里连 redis 这个主机名就是 Redis 容器
• 数据卷（volume）：容器删了，容器里写的文件就没了；数据卷是 Docker 管理的一块持久存储，挂到容器里，数据独立于容器保存

实验里三个容器的关系和启动顺序：

flowchart TB
    H["本机 localhost:8088"] --> N["nginx 容器<br/>监听 80"]
    subgraph NET["compose 内部网络，服务名就是主机名"]
        N -->|"proxy_pass http://api:8000"| A["api 容器<br/>uvicorn 2 个 worker<br/>以 appuser 运行"]
        A -->|"REDIS_HOST=redis"| R["redis 容器<br/>开 AOF"]
    end
    R --- V[("数据卷 redis-data")]
    R -. "1. 健康后" .-> A
    A -. "2. 健康后" .-> N

动手实验 9：新建一个目录 deploy/，放下面几个文件。

app.py：

import asyncio
import os
import threading
import uuid

import redis.asyncio as redis
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()
r = redis.Redis(host=os.environ["REDIS_HOST"], decode_responses=True)


@app.get("/health")
async def health():
    await r.ping()
    return {"ok": True}


@app.get("/visits")
async def visits():
    return {"visits": await r.incr("visits"), "worker_pid": os.getpid()}


@app.get("/stream")
async def stream():
    async def events():
        for i in range(1, 4):
            yield f"data: step {i}\n\n"
            await asyncio.sleep(1)
    return StreamingResponse(events(), media_type="text/event-stream")


@app.get("/slow")
async def slow(heartbeat: int = 0):
    async def events():
        yield "data: 开始编译\n\n"
        for _ in range(5):
            await asyncio.sleep(1)
            if heartbeat:
                yield ": ping\n\n"
        yield "data: 编译完成\n\n"
    return StreamingResponse(events(), media_type="text/event-stream")


local_lock = threading.Lock()


@app.get("/build-local")
async def build_local():
    if not local_lock.acquire(blocking=False):
        raise HTTPException(409, "已经有构建在跑")
    try:
        await asyncio.sleep(1)
        return {"started_by_pid": os.getpid()}
    finally:
        local_lock.release()


RELEASE = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0"


@app.get("/build-redis")
async def build_redis():
    token = uuid.uuid4().hex
    if not await r.set("lock:build", token, nx=True, ex=600):
        raise HTTPException(409, "已经有构建在跑")
    try:
        await asyncio.sleep(1)
        return {"started_by_pid": os.getpid()}
    finally:
        await r.eval(RELEASE, 1, "lock:build", token)

• redis.asyncio：redis-py 的异步版本，所有命令都要 await。在 async def 接口里必须用它，用同步客户端会卡住事件循环（14 篇 4.3 节）
• os.environ["REDIS_HOST"]：从环境变量读 Redis 地址，由 compose 传进来。配置和密钥都走环境变量，不写死在代码里
• os.getpid()：当前进程编号，用来看请求落在了哪个 worker 进程
• /slow 模拟一个中间 5 秒没有输出的编译步骤；heartbeat=1 时每秒发一行 : ping。SSE 里冒号开头的行是注释，浏览器会忽略，但对代理来说连接上有数据在流动
• /build-local 和 /build-redis 分别用进程内锁和 Redis 锁模拟”同时只允许一个构建”
• r.eval(脚本, 键的个数, 键..., 参数...)：直接执行一段 Lua 脚本，和 4.4 节的 register_script 是同一回事

requirements.txt，版本写死，保证每次构建装的一样：

fastapi==0.141.1
uvicorn==0.53.0
redis==8.1.0

Dockerfile：

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

RUN useradd --create-home appuser
COPY app.py .
USER appuser

EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

.dockerignore，构建时不发给 Docker 的文件：

.venv
__pycache__
*.db
.env

.env 必须排除，否则密钥会被打进镜像，任何能拉到镜像的人都能看到。

compose.yaml：

services:
  redis:
    image: redis:8-alpine
    command: ["redis-server", "--appendonly", "yes"]
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      timeout: 2s
      retries: 10

  api:
    build: .
    environment:
      REDIS_HOST: redis
    depends_on:
      redis:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
      interval: 5s
      timeout: 3s
      retries: 5

  nginx:
    image: nginx:1.29-alpine
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
    ports:
      - "8088:80"
    depends_on:
      api:
        condition: service_healthy

volumes:
  redis-data:

• --appendonly yes 加 redis-data 数据卷：开 AOF，数据写在卷里，容器重启数据还在
• healthcheck：Docker 定期执行 test 里的命令，成功就标记为健康。API 的健康检查调 /health，而 /health 里会 ping Redis，所以”API 健康”意味着它和 Redis 的连接也是通的
• depends_on 加 condition: service_healthy：等依赖的服务健康了才启动。只写 depends_on: [redis] 只保证先启动 Redis 容器，不保证 Redis 已经能接受连接（官方文档）
• API 没有写 ports，外面访问不到，只能经过 Nginx；只有 Nginx 把 80 端口映射到本机的 8088
• REDIS_HOST: redis：在 compose 的网络里，服务名就是主机名

nginx.conf，为了做对比实验写了四个路径，都转发到同一个 API：

server {
    listen 80;

    location /buffered/ {
        proxy_pass http://api:8000/;
    }

    location /gzip/ {
        proxy_pass http://api:8000/;
        gzip on;
        gzip_types text/event-stream;
    }

    location /short-timeout/ {
        proxy_pass http://api:8000/;
        proxy_buffering off;
        proxy_read_timeout 3s;
    }

    location /api/ {
        proxy_pass http://api:8000/;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_buffering off;
        proxy_read_timeout 300s;
    }
}

• proxy_pass http://api:8000/; 末尾带 /：把匹配到的前缀替换掉，/api/stream 转发成 /stream
• /buffered/：全部用默认值，也就是代理缓冲打开
• /gzip/：对事件流开 gzip 压缩。很多站点会在全局配置里对 text/* 类型统一开 gzip，事件流就被捎带上了
• /short-timeout/：把读超时从默认 60 秒缩到 3 秒，方便实验
• /api/：推荐的 SSE 配置。proxy_http_version 1.1 加清空 Connection 头，让 Nginx 和后端之间可以复用长连接

启动：

docker compose -p agentlab up -d --build

实际输出（Docker 27.5.1，截取最后几行）：

 Container agentlab-redis-1  Healthy
 Container agentlab-api-1  Starting
 Container agentlab-api-1  Started
 Container agentlab-api-1  Waiting
 Container agentlab-api-1  Healthy
 Container agentlab-nginx-1  Starting
 Container agentlab-nginx-1  Started

能看到启动顺序：Redis 健康之后 API 才启动，API 健康之后 Nginx 才启动。

验证 1：多 worker 和数据持久化。

for i in 1 2 3 4; do curl -s localhost:8088/api/visits; echo; done
docker compose -p agentlab restart redis
curl -s localhost:8088/api/visits
docker compose -p agentlab exec api whoami

{"visits":1,"worker_pid":9}
{"visits":2,"worker_pid":8}
{"visits":3,"worker_pid":8}
{"visits":4,"worker_pid":8}
{"visits":5,"worker_pid":8}
appuser

请求落在 8 号和 9 号两个进程上，计数是连续的，因为计数存在 Redis 里，所有进程共享；重启 Redis 后接着从 5 数，AOF 加数据卷生效了；容器里的进程以 appuser 身份运行。

验证 2：Nginx 后面的 SSE。 客户端脚本 sse_timing.py，记录每行数据到达的时间：

import time

import httpx


def watch(path):
    print(path)
    start = time.perf_counter()
    try:
        with httpx.stream("GET", "http://localhost:8088" + path, timeout=10,
                          headers={"Accept-Encoding": "gzip"}) as resp:
            for line in resp.iter_lines():
                if line:
                    print(f"  {time.perf_counter() - start:.2f}s 收到 {line}")
    except httpx.HTTPError as e:
        print(f"  {time.perf_counter() - start:.2f}s 连接出错 {type(e).__name__}")
    print(f"  {time.perf_counter() - start:.2f}s 流结束")


for path in ["/buffered/stream", "/gzip/stream", "/api/stream",
             "/short-timeout/slow", "/short-timeout/slow?heartbeat=1"]:
    watch(path)

• httpx.stream(...) 以流的方式发请求，resp.iter_lines() 每收到一行就交出一行，不等整个响应结束
• 请求头 Accept-Encoding: gzip 模拟浏览器，浏览器总是会带这个头
• Python 语法：except httpx.HTTPError as e。as e 把捕获到的异常对象存进变量 e，type(e).__name__ 取它的类型名

实际输出（nginx 1.29-alpine）：

/buffered/stream
  0.05s 收到 data: step 1
  1.06s 收到 data: step 2
  2.06s 收到 data: step 3
  3.07s 流结束
/gzip/stream
  3.06s 收到 data: step 1
  3.06s 收到 data: step 2
  3.06s 收到 data: step 3
  3.06s 流结束
/api/stream
  0.03s 收到 data: step 1
  1.03s 收到 data: step 2
  2.04s 收到 data: step 3
  3.05s 流结束
/short-timeout/slow
  0.03s 收到 data: 开始编译
  3.04s 连接出错 RemoteProtocolError
  3.04s 流结束
/short-timeout/slow?heartbeat=1
  0.03s 收到 data: 开始编译
  1.04s 收到 : ping
  2.05s 收到 : ping
  3.05s 收到 : ping
  4.06s 收到 : ping
  5.07s 收到 : ping
  5.07s 收到 data: 编译完成
  5.07s 流结束

前三个路径里三条事件的到达时间画成折线（横轴是第几条事件，纵轴是到达的秒数）。两条几乎重合、斜着上升的是默认缓冲和关闭缓冲，事件每秒到一条；平的那条是 gzip，三条都在第 3 秒才到：

xychart-beta
    title "三条事件的到达时间（秒）"
    x-axis ["第 1 条", "第 2 条", "第 3 条"]
    y-axis "到达时间（秒）" 0 --> 3.5
    line [0.05, 1.06, 2.06]
    line [3.06, 3.06, 3.06]
    line [0.03, 1.03, 2.04]

三个结论，其中第一个和很多文章（包括我自己 14 篇的写法）说的不一样：

1. 默认开着代理缓冲，事件也是实时到达的。在这个环境里，每条事件都在后端发出后几十毫秒内到了。Nginx 文档对缓冲的描述是”尽快从后端读取响应存进缓冲区”，并没有说要攒满才发给客户端。所以”Nginx 默认缓冲会让 SSE 卡住”不是在所有情况下都成立。我还是建议显式关掉缓冲（proxy_buffering off，或者后端返回响应头 X-Accel-Buffering: no，我项目就加了这个头）：行为不依赖版本、缓冲区大小和其他模块，而且写明了意图
2. gzip 才是真正会攒数据的。开了 gzip 以后，三条间隔一秒的事件在第 3 秒一起到达，用户看到的就是”卡了三秒突然全出来”。压缩器要攒够一定数据才输出一块。事件流不要压缩，检查全局配置里的 gzip_types 有没有把 text/event-stream 带进去
3. 读超时是两次读取之间的间隔。proxy_read_timeout 设成 3 秒，后端 5 秒没发数据，第 3 秒连接就被 Nginx 断开，最后的”编译完成”永远到不了；每秒发一行心跳注释，同样的 3 秒超时就能完整跑完。默认 60 秒，我项目里一条编译命令就可能超过 60 秒没有任何输出，所以心跳是必须补的

验证 3：多 worker 下的锁。 lock_workers.py 同时发 6 个请求：

import asyncio

import httpx


async def main():
    async with httpx.AsyncClient(base_url="http://localhost:8088/api") as client:
        for path in ["/build-local", "/build-redis"]:
            results = await asyncio.gather(*(client.get(path) for _ in range(6)))
            ok = [resp.json()["started_by_pid"] for resp in results if resp.status_code == 200]
            print(f"{path}: 同时发 6 个请求，{len(ok)} 个开始了构建，所在进程 {ok}，"
                  f"{sum(resp.status_code == 409 for resp in results)} 个返回 409")


asyncio.run(main())

跑了 3 次，结果一致：

/build-local: 同时发 6 个请求，2 个开始了构建，所在进程 [9, 8]，4 个返回 409
/build-redis: 同时发 6 个请求，1 个开始了构建，所在进程 [8]，5 个返回 409
/build-local: 同时发 6 个请求，2 个开始了构建，所在进程 [9, 8]，4 个返回 409
/build-redis: 同时发 6 个请求，1 个开始了构建，所在进程 [9]，5 个返回 409
/build-local: 同时发 6 个请求，2 个开始了构建，所在进程 [9, 8]，4 个返回 409
/build-redis: 同时发 6 个请求，1 个开始了构建，所在进程 [8]，5 个返回 409

进程内锁每个 worker 一把，2 个 worker 就放进来 2 个构建；Redis 锁所有进程共用一把，只放进来 1 个。

flowchart LR
    subgraph L["进程内锁：放进 2 个"]
        direction TB
        Q1[6 个并发请求] --> P8["worker 8<br/>自己的 threading.Lock"]
        Q1 --> P9["worker 9<br/>自己的 threading.Lock"]
        P8 --> OK1[1 个构建]
        P9 --> OK2[1 个构建]
    end
    subgraph RL["Redis 锁：放进 1 个"]
        direction TB
        Q2[6 个并发请求] --> R8[worker 8]
        Q2 --> R9[worker 9]
        R8 --> RK[("Redis 里同一个 lock:build")]
        R9 --> RK
        RK --> OK3["1 个构建，其余 409"]
    end
    L ~~~ RL

这就是 CrossBuild 现在的 threading.Lock 在 --workers 2 下会出的问题。

• Python 语法：sum(条件 for ...)。生成器表达式里每个元素是 True 或 False，sum 把它们当 1 和 0 加起来，就是满足条件的个数
• 两个 f-string 挨着写会自动拼接成一个字符串，用来把长字符串拆成两行

实验做完记得清理：docker compose -p agentlab down -v，-v 会连数据卷一起删，只在实验环境这么用。

生产部署还要考虑的：

4.11 线上排查 Python 服务

口述见 3.13。这一节是命令清单，没有做实验，每条命令的输出格式在不同系统上略有差别。

AI 应用额外常见的几种：模型调用没设超时，请求一直挂着把线程或连接占满；在 async def 里调了同步客户端，所有请求一起变慢（14 篇 4.3 节的实验）；对话历史或 Agent 轨迹一直保存在进程内存里，内存持续上涨。

第五部分 对照项目

CrossBuild 改造步骤

改造前后的结构对比。现在：

flowchart LR
    B[浏览器] -->|"GET /api/build 直接启动"| P
    subgraph P["uvicorn 单进程"]
        L["threading.Lock"]
        T["后台线程跑 Agent"] --> Q["queue.Queue"]
        Q --> S["SSE 只有 data:，无心跳"]
    end
    T --> DB[("SQLite")]
    T --> M[模型 API]

改完第 1 到 4 步之后：

flowchart TB
    B[浏览器] --> N["Nginx<br/>关缓冲、不 gzip"]
    N -->|"POST /api/builds"| A["API 多 worker"]
    N -->|"GET /api/builds/{id}/events<br/>带 Last-Event-ID"| A
    A -->|"Redis 锁、XADD 任务"| R[("Redis")]
    R -->|"消费组"| W["worker 容器<br/>跑 Agent，看门狗续锁"]
    W -->|"XADD 事件"| R
    R -->|"XREAD BLOCK，15 秒无事件发心跳"| A
    W --> DB[("SQLite，run_id 加索引")]
    W -->|"熔断 + 备用模型"| M[模型 API]

按”每一步都能单独验证、单独讲”的顺序排。第 1、2、3 步做完，面试时后端部分就有可讲的实物了。

第 1 步：docker-compose 跑起来（先不改代码）

• API 镜像里除了 Python 依赖，还要装 git、cmake、make 和 zig 工具链
• 知识库的 bge-m3 向量模型有几个 GB，不要打进镜像，用数据卷挂载模型缓存目录，或者容器里默认只用 BM25 检索（KB_MODE=bm25）
• runs.db、workspace/ 挂数据卷；DEEPSEEK_API_KEY 从 .env 传入，.dockerignore 排除 .env
• 验证：docker compose up 之后，浏览器里完整跑一次 libpng，历史记录里能看到

第 2 步：进程内锁换成 Redis 锁

• 接口里用 SET lock:build 随机值 NX PX 60000 抢锁，抢不到返回 409
• 随机值交给后台线程，在 worker 的 finally 里用比对脚本释放（保持 14 篇修过的”锁跟着任务走、不跟着连接走”）
• 构建一次要几分钟，加看门狗每 20 秒续期一次
• 验证：--workers 2 启动，用 4.10 节的并发脚本同时发 6 个请求，只有 1 个开始构建；把原来的 ServerLock 回归测试改成针对 Redis 锁

第 3 步：SSE 事件带编号、可续传、有心跳

• 后台线程把事件 XADD 进 events:run:{id}，同时设过期时间（比如 1 天）
• SSE 接口用 XREAD BLOCK 15000 读新事件，读到就发 id: 加 data:；15 秒没有新事件就发一行 : ping
• 浏览器重连时带 Last-Event-ID，从下一条开始补发
• 验证：构建进行中刷新页面，能接着看到后面的步骤；Nginx 读超时设 30 秒，长时间编译的步骤不会断

第 4 步：任务队列和独立工作进程

• 接口改成 POST /api/builds，写进 tasks:build 流，返回任务编号
• 新增一个 worker 服务，用消费组读任务、执行、XACK；启动时用 XAUTOCLAIM 认领上次崩溃留下的任务
• 验证：构建过程中重启 API 容器，构建不受影响；构建过程中杀掉 worker 容器，重启后任务被重新认领

第 5 步（可选）：API key 和每日配额

• 单用户部署，一个从环境变量读的 API key 就够；多用户按 4.7 节的表结构
• 每个 key 每天的构建次数和 token 用量用 4.5 节的脚本控制

做了才能写进简历。第 2、3 步做完，简历里可以写一句类似”用 Redis 分布式锁替换进程内锁，支持多 worker 部署；SSE 事件存入 Redis Streams，支持断线续传和心跳保活”，面试时能讲出 4.4 节和 4.10 节的实验对比。

第六部分 追问清单

第七部分 闭卷自测

1. Redis 说是单线程，为什么还能这么快？单线程带来什么要注意的地方？

2. 分别说出缓存穿透、击穿、雪崩的原因和解决办法。

3. 旁路缓存模式下，写数据时为什么是”先更新数据库、再删除缓存”？换成”先删缓存再更新数据库”会有什么问题？

4. 写出 Redis 分布式锁的加锁命令，说明释放时为什么不能直接 DEL。

5. 任务执行时间可能超过锁的过期时间，怎么办？这种锁在什么情况下仍然不可靠？

6. 固定窗口限流的边界问题是什么？滑动窗口怎么用 Redis 实现？

7. 用 Redis List 做任务队列有什么问题？Streams 消费组是怎么解决的？还剩什么问题要应用自己处理？

8. 401 和 403 有什么区别？在 FastAPI 里怎么复用鉴权逻辑？

9. JWT 的载荷是加密的吗？解码时为什么要指定 algorithms？JWT 怎么提前作废？

10. 用户密码和 API key 分别应该怎么存？为什么不一样？

11. 30 万行的工具调用表，按 run_id 查一次运行的调用很慢。怎么确认原因、怎么改？改完后 EXPLAIN 会有什么变化？

12. 下面两条 SQL 各有什么问题，怎么改？（1）WHERE DATE(started_at) = '2026-03-01'（2）ORDER BY id LIMIT 280000, 20

13. 熔断器有哪三种状态，怎么转换？它和重试有什么区别？

14. docker-compose 里 API 依赖 Redis，只写 depends_on: [redis] 够吗？uvicorn 开 2 个 worker 后，项目里哪些东西会出问题？

15. SSE 部署在 Nginx 后面，要注意哪几件事？你实测的结果是什么？

延伸阅读

1. Redis：Distributed locks — 单实例锁的正确写法和 Redlock
2. Martin Kleppmann：How to do distributed locking — 为什么基于过期时间的锁不能保证绝对正确，fencing token
3. Redis：Streams — 消费组、待确认列表、XAUTOCLAIM
4. Redis：Key eviction — 淘汰策略、近似 LRU
5. Redis：SET 命令 — NX、PX 等选项
6. FastAPI：OAuth2 with Password, Bearer with JWT tokens — 官方的 PyJWT 加 pwdlib 示例
7. OWASP：Password Storage Cheat Sheet — Argon2id、scrypt、bcrypt 的参数要求
8. RFC 7519：JSON Web Token — JWT 的标准定义
9. MySQL：EXPLAIN Output Format — 每一列和 Extra 各种取值的含义
10. Use The Index, Luke：No Offset — 为什么不该用偏移分页
11. 微软：Circuit Breaker pattern — 熔断器的状态和设计考虑
12. Docker：Building best practices — 层缓存、多阶段构建、非 root 用户
13. Docker Compose：Control startup order — depends_on 和健康检查
14. Nginx：ngx_http_proxy_module — proxy_buffering、proxy_read_timeout

以上链接在 2026-09-19 打开核对过。

下一篇：15 系统设计题——系统设计里的缓存、队列、限流、鉴权，用的就是这一篇的内容。