《AI时代的程序员修养》前两篇讲了程序和数据结构。这一篇进入另一个很容易被 AI 编程掩盖的问题：代码到底由谁执行，在哪里执行，什么时候让出 CPU，什么时候会把整个服务卡住。

很多并发 bug 不是因为代码太难，而是因为执行模型一开始没有说清楚。你让 AI “并发处理一批任务”，它可能写线程池；你让它“改成异步”，它可能只是在函数前面加 async；你让它“提高吞吐”，它可能多开几个 worker，却没有发现数据库连接池已经被打满。

执行模型不是底层八股。它决定系统能不能扛住慢接口、突发流量、CPU 计算、文件读写和外部服务抖动。

并发不是一个词

“并发执行”这句话太粗了。

它可能表示多个独立进程同时跑，也可能表示一个进程里多个线程抢时间片，还可能表示一个线程里用协程在等待 I/O 时切来切去。它们看起来都能“同时处理多个任务”，但成本、隔离性、共享状态和故障模式完全不同。

执行模型	常见场景	优点	风险
多进程	Web worker、批处理 worker、CPU 密集任务	隔离强，单进程崩溃影响小，能利用多核	内存占用高，进程间通信成本高
多线程	阻塞 I/O 并发、后台任务、GUI 程序	共享内存方便，切换成本低于进程	竞态、死锁、线程池耗尽
协程 / async	高并发网络 I/O、API 网关、爬取、长连接	单线程可承载大量连接，等待时切换便宜	一个阻塞调用就能卡住事件循环
事件驱动	消息消费、前端 UI、Node.js 服务、异步框架	适合大量短小事件和 I/O 回调	调用链分散，错误传播和上下文追踪复杂
批处理	数据同步、报表、离线任务	吞吐高，容易做批量优化	延迟高，失败恢复和断点续跑要设计

AI 很容易把这些模型混在一起。比如在 FastAPI 里生成一个 async def 接口，然后里面调用同步数据库驱动、同步 HTTP 客户端或者本地大文件读取。代码语法上是异步的，运行时却会把事件循环堵死。

所以写 prompt 时不要只说“异步处理”，要说清楚：

这是一个 async Web 接口，运行在单进程事件循环中。
请求路径不能执行阻塞 I/O。
数据库访问必须使用 async driver。
CPU 密集计算必须 offload 到进程池或独立 worker。
外部 HTTP 调用必须设置超时和并发上限。

这几句话比“帮我优化性能”有用得多。

进程：隔离很贵，但很可靠

进程是操作系统给程序划出的隔离空间。每个进程有自己的地址空间、文件描述符、环境变量、堆栈和运行状态。一个进程崩溃，通常不会直接踩坏另一个进程的内存。

Web 服务里常见的多进程模型，是一个 master 拉起多个 worker：

master
  ├── worker-1
  ├── worker-2
  ├── worker-3
  └── worker-4

每个 worker 都能接请求。对 Python、Node.js 这类运行时来说，多进程也是绕开单进程 CPU 限制、利用多核的常见办法。

但进程不是免费的。

如果一个模型加载到内存需要 2GB，开 8 个进程可能就不是 2GB，而是接近 16GB，具体还取决于 copy-on-write、模型是否被修改、运行时内存分配方式。数据库连接也类似。每个进程一个连接池，4 个 worker、每个池 20 条连接，总连接数就是 80，不是 20。

让 AI 配 Web worker 时，很容易得到这种建议：

CPU 是 8 核，所以 worker 数设为 8。

这句话只对一小部分场景成立。更合理的估算要看：

• 单个 worker 的常驻内存。
• 单个 worker 的数据库连接池大小。
• 请求是 CPU 密集还是 I/O 密集。
• 单机是否还跑着其他服务。
• 下游数据库、缓存、第三方 API 能不能承受并发。
• worker 重启时是否会同时触发冷启动洪峰。

一个更靠谱的 AI 指令可以这样写：

请不要只按 CPU 核数推荐 worker 数。
先估算：
1. 单 worker 常驻内存。
2. 每个 worker 的 DB 连接池大小。
3. 峰值请求并发。
4. 下游数据库最大连接数。
5. CPU 密集逻辑占比。
然后给出 worker 数、连接池大小和压测验证方法。

进程模型的关键词是隔离。隔离带来可靠性，也带来资源放大。AI 如果只看代码，不看部署资源，就很容易把“能跑”配成“上线就挤爆”。

线程：共享内存方便，也最容易出暗伤

线程在同一个进程里共享内存。共享内存让通信变简单，也让错误更隐蔽。

比如一个很常见的计数器：

count = 0


def handle_request():
    global count
    count += 1

在单线程里它没问题。在多线程里，count += 1 不是原子操作。它至少包含读取、加一、写回。两个线程同时执行，就可能丢计数。

AI 很容易生成这种代码，因为它在单元测试里通常能过。测试跑 100 次也许都没问题，线上高并发时才偶发。

线程问题最麻烦的地方就在这里：bug 不稳定，复现困难，日志看起来也不像错。

常见线程风险包括：

• 多线程同时修改同一个 dict、list 或对象字段。
• 一个线程持有锁后调用慢 I/O，导致其他线程长期等待。
• 多把锁顺序不一致，形成死锁。
• 线程池里又提交任务等待同一个线程池执行，造成线程饥饿。
• request 上下文、用户身份、trace id 放在线程局部变量里，但切换执行方式后丢失。

线程池尤其容易被 AI 用坏。比如把同步外部 API 调用丢进线程池：

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=100)


def call_all(items):
    futures = [executor.submit(call_remote_api, item) for item in items]
    return [future.result() for future in futures]

这段代码看起来很会并发，实际缺少几个关键约束：

• items 如果有 10 万个，会一次性提交 10 万个 future。
• 每个调用没有超时，线程可能一直占着。
• 没有失败隔离，一个慢下游拖住整个批次。
• 没有全局限流，多个请求同时进来会把线程池打满。
• 没有 backpressure，调用方不知道系统已经忙不过来。

更稳的实现要控制批量、超时和并发：

from concurrent.futures import ThreadPoolExecutor, as_completed


def call_bounded(items, *, max_workers=20, batch_size=200, timeout=3):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        for start in range(0, len(items), batch_size):
            batch = items[start:start + batch_size]
            futures = [executor.submit(call_remote_api, item) for item in batch]
            for future in as_completed(futures, timeout=timeout * len(batch)):
                results.append(future.result(timeout=timeout))
    return results

这段代码仍然只是示意，线上还要处理异常、取消、重试、指标和部分失败。但它已经把线程池从“无限提交”改成了“受控并发”。

让 AI 写线程代码时，prompt 里一定要出现这些词：共享状态、锁顺序、线程池大小、队列长度、超时、取消、限流、指标。

协程：不阻塞才有意义

协程的价值不在于它神奇，而在于它在等待 I/O 时可以主动让出执行权。

一个 async 服务大概是这样工作的：

event loop
  ├── request A 等数据库
  ├── request B 等 HTTP 响应
  ├── request C 正在写 socket
  └── request D 准备返回

当 A 等数据库时，事件循环可以去处理 B、C、D。这个模型能用少量线程承载大量连接。

前提是：等待必须是可让出的等待。

下面这段代码语法上是 async，行为上却可能是灾难：

import requests


async def fetch_user(user_id: str):
    response = requests.get(f"https://example.com/users/{user_id}")
    return response.json()

requests.get 是同步阻塞调用。在它返回前，事件循环不能去处理其他请求。并发一上来，整个服务延迟都会被拖高。

应该使用异步 HTTP 客户端：

import httpx


async def fetch_user(user_id: str):
    timeout = httpx.Timeout(3.0, connect=1.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(f"https://example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

数据库也是一样。async def 里面如果用同步 ORM 查询，仍然会阻塞。文件读写、大 JSON 解析、图片处理、压缩、加密、大量正则匹配，也都可能让事件循环卡住。

检查 async 代码时，我通常先看这几类危险点：

• requests、同步数据库驱动、同步 SDK。
• time.sleep()，而不是 await asyncio.sleep()。
• 大文件同步读写。
• CPU 密集循环。
• 没有超时的外部调用。
• asyncio.gather() 一次性提交过多任务。
• 忘记 await，导致协程对象没有执行。

AI 写 async 代码时，最容易把 gather 当万能并发：

results = await asyncio.gather(*[fetch(item) for item in items])

如果 items 有 5 万个，这就是一次性创建 5 万个协程。即使每个协程只是网络 I/O，下游服务、DNS、连接池和本机内存也可能扛不住。

更好的写法是加并发上限：

import asyncio


async def gather_limited(items, limit: int = 100):
    semaphore = asyncio.Semaphore(limit)

    async def run_one(item):
        async with semaphore:
            return await fetch(item)

    return await asyncio.gather(*(run_one(item) for item in items))

如果结果可以流式处理，还可以不要一次性攒全部结果。并发不是越大越好，并发必须被上限、超时和下游容量约束。

CPU 密集和 I/O 密集要分开想

很多“性能优化”的误判来自把 CPU 密集和 I/O 密集混在一起。

I/O 密集任务大部分时间在等：等数据库、等网络、等磁盘、等第三方 API。协程和线程都可能有帮助，关键是不要把等待时间浪费成空转。

CPU 密集任务大部分时间在算：压缩、加密、图片处理、模型推理、复杂解析、大量排序、矩阵计算。协程对它帮助很小，因为它一直占着 CPU，不会自然让出事件循环。

一个简单判断方法：

现象	更可能是什么	优先考虑
CPU 很低，延迟很高	I/O 等待、锁等待、连接池等待	超时、连接池、异步 I/O、批量、下游排查
CPU 打满，队列堆积	CPU 密集计算	进程池、独立 worker、算法优化、批处理
DB 连接耗尽	并发过高或连接泄漏	池大小、请求上限、慢查询、事务边界
事件循环延迟升高	async 中有阻塞或 CPU 计算	找同步调用、offload、拆 worker
线程数很多但吞吐不涨	下游瓶颈或锁竞争	限流、锁粒度、批量、降级

假设一个 API 需要上传 CSV、解析 20 万行、写数据库。AI 可能直接在请求里写：

async def upload(file):
    rows = parse_csv(file)
    await insert_rows(rows)
    return {"ok": True}

这会让请求链路背上全部成本。解析可能阻塞事件循环，写入可能占用长事务，用户连接也要一直等。

更像工程系统的设计应该是：

1. 请求只做鉴权、保存文件、创建任务。
2. 后台 worker 读取任务，分块解析。
3. 每批数据校验后批量写入。
4. 任务状态可查询，失败可定位到批次。
5. 对大文件设置大小、行数和耗时限制。
6. worker 有并发上限，数据库写入有批量和事务边界。

这里的执行模型已经从“一个 async 请求”变成“请求 + 队列 + worker + 批处理”。这不是代码风格问题，而是系统边界问题。

阻塞点要写在设计里

AI 最难凭空猜到的是阻塞点。

阻塞点包括但不限于：

• 数据库查询和事务。
• 外部 HTTP/RPC 调用。
• 文件系统读写。
• 日志写入。
• 消息队列生产和消费。
• 锁等待。
• CPU 计算。
• 第三方 SDK 内部的同步调用。
• DNS、TLS 握手、连接池排队。

如果 prompt 只写“实现接口”，AI 默认会把这些东西塞在最顺手的位置。你需要明确告诉它哪些能在请求内做，哪些必须异步化，哪些要限流，哪些要超时。

比如一个生成报告的接口，可以这样描述：

实现“生成报告”功能，但先不要写代码。

执行模型要求：
- HTTP 请求只创建 report_job，不能同步生成完整报告。
- 生成过程由后台 worker 执行。
- 单个 worker 同时最多生成 2 个报告。
- 数据库查询每次最多读取 5000 行，必须分页。
- 调用外部服务必须设置 10 秒超时，最多重试 2 次。
- 报告生成状态包括 pending/running/success/failed/cancelled。
- 用户可以查询进度，也可以取消 pending/running 的任务。
- worker 崩溃后，running 超过 30 分钟的任务可以被重新捞起。

这段话没有写一行实现，但已经把执行模型定住了。AI 接下来写出来的代码才不会把一个 5 分钟任务塞进 HTTP 请求。

共享状态要尽量少

执行模型一旦涉及并发，共享状态就是风险源。

共享状态不一定是全局变量。数据库行、缓存 key、本地文件、消息 offset、任务状态、用户额度、连接池、临时目录，都可能是共享状态。

一个典型例子是“同一个任务只能运行一次”。如果只在代码里判断：

job = get_job(job_id)
if job.status == "running":
    return
job.status = "running"
save(job)
run(job)

两个 worker 同时读到 pending，就可能都开始执行。

正确的核心不是 Python 代码，而是原子抢占：

update jobs
set status = 'running',
    locked_by = :worker_id,
    locked_at = now()
where id = :job_id
  and status = 'pending';

然后检查受影响行数。只有更新成功的 worker 才能执行任务。

这就是执行模型和数据结构结合的地方：并发控制不能靠“我觉得不会同时发生”，要靠数据库原子操作、唯一约束、锁、版本号或幂等键。

让 AI 写这类代码时，可以直接要求：

不能用“先查再改”的方式抢任务。
必须使用单条原子 UPDATE 或 SELECT ... FOR UPDATE SKIP LOCKED。
请写出并发下两个 worker 同时抢同一任务的测试用例。

最后一句很重要。只有让 AI 写并发测试，它才更可能意识到这不是普通 CRUD。

事件循环延迟是 async 服务的体温

如果一个 async 服务偶尔突然变慢，不一定是数据库慢，也不一定是机器不够。可能是事件循环被阻塞了。

事件循环延迟可以理解为：本来某个协程应该在 10 毫秒后被调度，结果 300 毫秒后才轮到它。说明这段时间事件循环忙着做别的事，或者被某个同步调用卡住了。

可以用一个简单后台任务监控：

import asyncio
import time


async def monitor_event_loop_lag(interval: float = 0.1):
    while True:
        start = time.perf_counter()
        await asyncio.sleep(interval)
        lag = time.perf_counter() - start - interval
        if lag > 0.05:
            record_metric("event_loop_lag_seconds", lag)

这段代码不能解决问题，但能让问题被看见。

当事件循环延迟升高时，要重点查：

• 最近是否加入了同步 SDK。
• 是否有大 JSON 序列化或反序列化。
• 是否在请求里做了图片、压缩、加密、复杂正则。
• 是否有日志同步写入慢存储。
• 是否一次性 gather 了太多任务。
• 是否把大量数据一次性加载到内存再处理。

AI 生成的 async 服务，如果没有事件循环延迟、请求耗时、下游耗时、连接池等待这些指标，就很难判断它到底在慢什么。

一份给 AI 的执行模型提示词

以后让 AI 写涉及并发、异步、后台任务的代码，可以先让它回答这组问题：

先不要写代码。

请为这个功能设计执行模型：
1. 哪些逻辑运行在 HTTP 请求内，哪些运行在后台 worker。
2. 哪些操作是 I/O 密集，哪些是 CPU 密集。
3. 是否使用进程、线程、协程、消息队列或批处理，为什么。
4. 列出所有可能阻塞事件循环或线程池的调用。
5. 给出并发上限、队列长度、超时、重试、取消和降级策略。
6. 列出共享状态，以及每个共享状态如何保证并发安全。
7. 给出数据库连接池、HTTP 连接池和 worker 数量的估算方式。
8. 设计至少 5 个测试：并发抢占、超时、取消、下游慢、worker 崩溃恢复。
9. 最后再给代码实现计划。

这段 prompt 的目的不是把 AI 变成架构师，而是逼它先暴露假设。假设摆出来，人才能判断。

如果 AI 的回答里没有“阻塞点”“共享状态”“超时”“连接池”“并发上限”“失败恢复”，那它大概率还停留在 demo 代码层面。

最后看几个验收问题

执行模型写完之后，不要急着相信代码。至少问一遍：

• 这个接口里有没有同步 I/O？
• async 函数里有没有 CPU 密集计算？
• 线程池、连接池、任务队列有没有上限？
• 多 worker 会不会执行同一个任务？
• 外部调用有没有超时、重试和熔断边界？
• 请求取消后，后台任务会不会继续占资源？
• worker 崩溃后，running 状态如何恢复？
• 日志、指标、trace 能不能看出卡在哪里？
• 压测时 CPU、内存、DB 连接、事件循环延迟分别是什么表现？

这些问题都不花哨，但它们决定系统能不能从“本地可运行”走到“线上可解释”。

AI 会越来越擅长生成并发代码。可并发代码最贵的部分，从来不是把语法写对，而是知道什么东西应该同时发生，什么东西绝对不能同时发生，什么东西慢了必须让路，什么东西失败后必须恢复。

这就是程序员不能交给 AI 猜的执行模型。

微信公众号

欢迎关注「字与码」

如果这篇文章对你有用，也欢迎在微信里继续关注后续更新。

X / Twitter

关注 @ax2_zicode

更即时的技术观察、新文章提醒和一些短想法会发在 X 上。

X 打开 X 主页 @ax2_zicode