写过 Go 的人学 Python 协程，最容易踩的坑是把 await 当成 go，把 asyncio 当成 Go runtime。两者都叫“协程”，但调度方式、编程模型和出错方式差别很大。
先给结论：
- Python 的协程是协作式的：只有遇到
await才会让出执行权，没有await就没人能打断它。 - Go 的 goroutine 由 runtime 隐式调度，并且支持抢占，写代码时几乎感知不到切换点。
- Python 把“能暂停”写进了函数签名（
async def），代价是同步与异步代码之间有一道墙；Go 没有这道墙。
下文假设你熟悉 goroutine、channel 和 context，只讲 Python 这一侧的机制，再逐项对照。示例在 Python 3.10 下验证过。
语法与运行时两层
async 和 await 是 Python 语言本身的关键字（3.5 起），不属于 asyncio。asyncio 是标准库里运行它们的引擎。两者是不同的层：
| 层 | 是什么 | 例子 |
|---|---|---|
| 语法 | 语言关键字 | async def、await、async with、async for |
| 运行时 | 事件循环，负责调度协程 | asyncio，也可以是 trio、anyio |
语法只定义“这个函数可以在等待时暂停”，由谁来恢复它，要靠事件循环。
这和 Go 形成第一个对比：Go 的 go 关键字和调度器同属语言运行时，用户无法替换；Python 的语法与运行时是解耦的，所以同一套 async/await 可以跑在不同的事件循环上。
flowchart TD
A["async def / await<br/>语言语法"] --> B["事件循环"]
B --> C["asyncio"]
B --> D["trio"]
B --> E["uvloop"]
四个关键字的含义：
async def f()：定义协程函数。await x：等待x完成，等待期间当前协程暂停，事件循环去跑别的协程。async with：进入和退出时需要await的上下文管理器，比如从连接池借连接要等池里有空位。async for：每取下一项都可能要等待的循环，比如流式读取模型输出：async for chunk, _ in agent.astream(...)。
async with 与 async for
这两个关键字是普通 with、for 的异步版本，区别在于底层协议里的方法本身是协程，可以挂起。
普通 with 调用 __enter__ / __exit__，不能暂停。但有些资源的进入和退出要等 IO：借连接时池已满，要等别人归还；归还时要等网络往返。写成同步方法会卡死事件循环，所以改用 __aenter__ / __aexit__：
import asyncio
class Pool:
def __init__(self, size):
self._sem = asyncio.Semaphore(size) # 模拟连接池容量
async def __aenter__(self):
await self._sem.acquire() # 池满时挂起，等别人归还
return "conn"
async def __aexit__(self, exc_type, exc, tb):
await asyncio.sleep(0.1) # 模拟归还时的网络往返
self._sem.release()
async def job(pool):
async with pool as conn: # 进入、退出时各可能挂起一次
await asyncio.sleep(0.2)
async def main():
pool = Pool(1) # 只有 1 个连接，两个任务会排队
await asyncio.gather(job(pool), job(pool))
asyncio.run(main())async for 同理，对应 __aiter__ / __anext__，每取下一项都可以 await。最简单的写法是异步生成器（带 yield 的 async def）：
async def ticks(n):
for i in range(n):
await asyncio.sleep(0.1) # 每次取下一项都要等
yield i
async def main():
async for i in ticks(3): # 等待期间事件循环可以跑别的协程
print("got", i)前面提到的 agent.astream(...) 就是这种异步迭代器：模型每吐出一个 chunk，循环体执行一次，等下一个 chunk 时协程挂起。
Go 里没有对应的区分：defer conn.Close() 里的 Close 本来就可以阻塞，只会阻塞当前 goroutine。Python 需要单独的 async with，是因为“等待”必须显式表示为“挂起”，这也是后面要讲的函数着色的一种体现。
协程对象
调用协程函数不会执行函数体，只返回一个协程对象：
import asyncio
async def fetch(name, delay):
await asyncio.sleep(delay)
return f"{name} done"
async def main():
c = fetch("a", 0.1) # 此时函数体还没开始执行
print(type(c)) # <class 'coroutine'>
print(await c) # await 才真正驱动它运行
asyncio.run(main()) # 创建事件循环并运行入口协程这是与 Go 最直观的差别：go f() 立刻把 f 交给调度器；f() 在 Python 里只是造了个对象。忘记 await 是最常见的 bug，Python 会给出 coroutine was never awaited 警告，函数体则根本没有执行。
协程对象本质上是一个可暂停、可恢复的状态机。每次 await 一个尚未完成的对象，协程就把控制权交还给事件循环，等对象就绪后再被恢复。
await 与 Task 的区别
直接 await 一个协程调用是串行的：当前协程等它跑完再往下走，并不会产生并发。要并发，必须把协程包成 Task，交给事件循环调度。下面沿用上一节的 fetch，对比两种写法：
import time
async def main():
t = time.perf_counter()
# 串行：第一个 await 结束，第二个才开始
await fetch("x", 0.3)
await fetch("y", 0.3)
print("串行", round(time.perf_counter() - t, 1)) # 约 0.6 秒
t = time.perf_counter()
# 并发：gather 把两个协程包成 Task，同时交给事件循环
r = await asyncio.gather(fetch("x", 0.3), fetch("y", 0.3))
print("并发", round(time.perf_counter() - t, 1)) # 约 0.3 秒
asyncio.run(main())对应关系：asyncio.create_task(fetch(...)) 近似于 go fetch(...)，不同点在于它返回一个可以 await、可以取消的 Task 句柄，而 Go 的 goroutine 没有句柄。
事件循环如何调度
事件循环是一个单线程循环，大致做三件事：
- 取出一个就绪的协程，运行到下一个
await。 - 该协程等待的对象（IO、定时器、Future）尚未完成时，把协程挂起，登记回调。
- 对象完成后，把协程重新放回就绪队列。
sequenceDiagram
participant L as 事件循环
participant A as 协程 A
participant B as 协程 B
L->>A: 运行
A-->>L: await 网络 IO（挂起）
L->>B: 运行
B-->>L: await sleep（挂起）
L->>L: 等待 IO 事件（epoll 等）
L->>A: IO 就绪，恢复
底层 IO 复用（Linux 上通常是 epoll）与 Go 的网络轮询器思路相同。区别在于：Go 的调度器把 goroutine 分发到多个线程（GMP 模型），而默认的 asyncio 事件循环只有一个线程，所有协程在同一个线程里轮流跑。
由此得到两个直接后果：
- 协程之间不存在真正的并行，CPU 密集任务不会因为
async变快。 - 两个
await之间的代码天然是原子的，不会被其他协程打断，因此很多场景不需要锁。
协作式与抢占式
这是两种模型最本质的差别。下面故意写一个错误示范，对比同步阻塞和异步等待：
import asyncio, time
async def blocking():
time.sleep(0.5) # 错误：同步阻塞，不让出执行权，整个事件循环被卡住
async def waiting():
await asyncio.sleep(0.5) # 正确：挂起当前协程，事件循环去跑别的
async def ticker():
for _ in range(3):
print("tick", round(time.perf_counter() % 100, 1))
await asyncio.sleep(0.1)
async def main():
await asyncio.gather(ticker(), blocking()) # 换成 waiting() 对比
asyncio.run(main())gather 按参数顺序调度，所以 ticker 先跑，第一次 tick 立刻打印，随后在 await sleep(0.1) 处让出执行权。接着轮到 blocking，它里面没有 await，time.sleep 占着线程不放，ticker 即使 0.1 秒的定时器早已到期，也要等 0.5 秒后才能被恢复，所以第二次 tick 约在 0.5 秒后才出现，随后两次紧挨着输出。换成 waiting 后，三次 tick 按 0.1 秒间隔正常输出。两者都“等了 0.5 秒”，区别是前者占着线程干等，后者把线程让给了别人。
这个现象和调度顺序有关：把参数顺序换成 gather(blocking(), ticker())，blocking 先拿到执行权，连第一次 tick 也会被推迟 0.5 秒。
同样的问题在 Go 里不会这样发生：goroutine 里写死循环，其他 goroutine 照样运行。Go 1.14 起引入基于信号的异步抢占，runtime 可以在函数调用之外的位置打断长时间运行的 goroutine。
| 维度 | Python asyncio | Go |
|---|---|---|
| 切换点 | 显式 await |
runtime 隐式决定 |
| 线程 | 默认单线程 | 多线程并行（GMP） |
| 长时间计算 | 卡死事件循环 | 可被抢占 |
| 数据竞争 | 两个 await 之间原子 |
需要锁或 channel |
代价对称：Go 需要考虑数据竞争，Python 需要考虑“有没有哪里悄悄阻塞了事件循环”。
函数着色问题
await 只能出现在 async def 里，同步函数无法直接调用协程并拿到结果。于是函数被分成两种“颜色”，异步会沿着调用链向上传染：底层一个函数改成 async，所有调用者都得跟着改。
def sync_caller():
result = fetch("a", 0.1) # 拿到的是协程对象，不是结果
# await fetch(...) # SyntaxError：await 只能在 async 函数内这个问题通常被称为“函数着色”。Go 没有这个问题：任何函数都可以放进 go，阻塞调用写起来和普通代码一样，调度由 runtime 负责。
反过来看，着色也有好处：函数签名告诉调用者“这里可能暂停”，代码审查时一眼能找到所有切换点。Go 里一个看似普通的函数调用，内部可能就阻塞在 channel 或锁上。
通信原语对照
| 需求 | Go | Python asyncio |
|---|---|---|
| 启动并发任务 | go f() |
asyncio.create_task(f()) |
| 等待一组任务 | sync.WaitGroup |
asyncio.gather / TaskGroup |
| 任务间传值 | channel | asyncio.Queue |
| 一次性结果 | 带缓冲 channel 或 chan T |
asyncio.Future / Task 本身 |
| 互斥 | sync.Mutex |
asyncio.Lock |
| 多路等待 | select |
asyncio.wait(..., return_when=FIRST_COMPLETED) |
Python 没有语言级的 channel，也没有 select 语句，通信靠 asyncio.Queue。下面用 Queue 实现 Go 里常见的 worker pool：
import asyncio
async def worker(q, results):
while True:
item = await q.get() # 队列空时挂起
try:
results.append(item * 2)
finally:
q.task_done() # 配合 q.join() 判断全部处理完
async def main():
q, results = asyncio.Queue(maxsize=2), []
workers = [asyncio.create_task(worker(q, results)) for _ in range(3)]
for i in range(10):
await q.put(i) # 队列满时挂起，等价于带缓冲 channel 的背压
await q.join() # 等所有任务处理完（worker 仍挂在 q.get() 上）
for w in workers:
w.cancel() # 只是发出取消请求，worker 还没真正退出
# 等 worker 真正退出；被取消的 task 以 CancelledError 结束，
# return_exceptions=True 把它当普通结果收集，避免异常打断 main
await asyncio.gather(*workers, return_exceptions=True)
print(sorted(results))
asyncio.run(main())注意两处与 Go 的差别：
- Queue 没有
close。Go 里close(ch)让接收方的range自然退出；Python 需要发送哨兵值或直接取消 worker。 - 没有
select。要同时等待多个来源，只能用asyncio.wait或wait_for组合。
取消与超时
Go 用 context 显式传递取消信号，每个可取消的函数都要接收 ctx 并监听 ctx.Done()。Python 把取消做进了 Task：对 Task 调用 cancel()，下一次它被恢复时，在挂起的那个 await 处抛出 asyncio.CancelledError。
async def slow():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
print("cleanup") # 做清理
raise # 必须重新抛出，否则取消被吞掉
async def main():
try:
await asyncio.wait_for(slow(), timeout=0.2)
except asyncio.TimeoutError:
print("timeout")
asyncio.run(main())| 维度 | Go context | Python CancelledError |
|---|---|---|
| 传递方式 | 显式参数，层层传递 | 隐式，沿 await 链传播 |
| 响应取消 | 主动检查 ctx.Done() |
在 await 处自动抛异常 |
| 忘记处理 | 取消无效，goroutine 泄漏 | 吞掉异常会让取消失效 |
| 超时 | context.WithTimeout |
asyncio.wait_for |
Python 的方式写起来省事，但有一个与协作式调度相关的盲区：取消只在 await 点生效。一个不含 await 的循环永远无法被取消。
CancelledError 在 3.8 起继承自 BaseException，所以 except Exception 不会误吞它；但 except BaseException 或裸 except: 会。
阻塞调用的处理
Go 里调用阻塞的系统调用或 CGO 时，runtime 会把线程交接出去，其他 goroutine 不受影响。Python 没有这种保护，同步阻塞调用（time.sleep、requests.get、同步数据库驱动）必须自己隔离，常见做法是放进线程：
async def good():
# 把同步阻塞函数丢进线程池，事件循环不被占用
await asyncio.to_thread(time.sleep, 0.5)asyncio.to_thread 从 3.9 开始提供。CPU 密集任务因为 GIL 的存在，线程帮不上忙，要用 ProcessPoolExecutor 配合 loop.run_in_executor。
实践上的更好选择是直接用异步生态的库：httpx/aiohttp 代替 requests，asyncpg 等异步驱动代替同步驱动。
LLM 流式调用场景
LLM 应用是 asyncio 最典型的用武之地：一次请求要等模型数秒甚至更久，期间连接一直挂着，每隔一小段时间吐出一个 chunk。绝大部分时间花在等网络上，CPU 几乎空闲。
import asyncio, time
async def astream(prompt):
# 模拟 LLM 流式输出：每个 chunk 之间都在等网络
for i in range(5):
await asyncio.sleep(0.1)
yield f"{prompt}-{i}"
async def handle(prompt):
parts = []
async for chunk in astream(prompt): # 每取一个 chunk 都可能挂起
parts.append(chunk) # 真实场景在这里推送给前端
return "".join(parts)
async def main():
t = time.perf_counter()
results = await asyncio.gather(*(handle(f"q{i}") for i in range(1000)))
print(len(results), round(time.perf_counter() - t, 1)) # 约 0.5 秒，而不是 500 秒
asyncio.run(main())1000 个并发流在一个线程里完成，总耗时约等于单个流的耗时。这里没有 GIL 的问题，原因有两点：
- 只有一个线程，不存在线程间争抢 GIL。
- 协程挂起时线程在等 IO 事件，没有在执行字节码。
所以“Python 做 IO 密集的高并发”成立，用 asyncio 加异步 SDK（如 httpx、各厂商的 async client）即可。
需要留意的是单核上限。每个 chunk 到达时要做反序列化、拼接、推送，这些 CPU 开销累计到吃满一个核，吞吐就到顶了。工程上的做法是多进程部署（如 uvicorn 多 worker），每个进程一个事件循环；同时避免在循环里做重计算，比如超大 JSON 解析或本地 embedding，这类任务放进进程池。
和 Go 对比：IO 并发能力在同一思路上，差别在于 Go 一个进程可以用满多核，Python 要靠多进程横向扩展。
选型建议
| 场景 | 建议 |
|---|---|
| 大量网络 IO、长连接、LLM 流式输出 | asyncio 很合适，吃满单核后多进程扩展 |
| CPU 密集计算 | 多进程；或换用 Go |
| 已有大量同步代码 | 线程池隔离，逐步迁移 |
| 需要多核并行的高并发服务 | Go 更自然 |
对 Go 开发者的几条心智转换：
- 把
await理解为“这里会让出执行权”，不是“启动并发”；并发由create_task或gather产生。 - 事件循环里禁止同步阻塞，这是 asyncio 里最重要的一条纪律。
- 取消靠异常传播，写
except时要小心别吞掉CancelledError。 - 不要期待 channel 与
select，用 Queue 加asyncio.wait组合。
总结
Python 协程把“何时切换”交给程序员，用 async/await 在代码里标出每个切换点，换来单线程下无需加锁的简单性；Go 把调度全部交给 runtime，换来更低的心智负担和多核并行，代价是需要自己处理数据竞争。理解了协作式调度这一条，其余差异（函数着色、取消只在 await 生效、阻塞调用要隔离）都是它的推论。
延伸阅读：Python 官方文档 asyncio 章节（Coroutines and Tasks、Event Loop）；Bob Nystrom 的 What Color is Your Function?；Go 1.14 发布说明中关于异步抢占的部分。