113 lines
4.8 KiB
Markdown
113 lines
4.8 KiB
Markdown
---
|
||
id: python.asyncio.coroutines-tasks
|
||
title: 协程、任务和事件循环：和 JS Promise 对照
|
||
type: lesson
|
||
domain: python
|
||
stack: [asyncio]
|
||
tags: [协程, 事件循环, Task, await, async for]
|
||
level: basic
|
||
prerequisites: [python.python.syntax-for-js]
|
||
est_hours: 2
|
||
related: [python.asyncio.concurrency-control, python.asyncio.interview]
|
||
week: [4]
|
||
order: 1
|
||
---
|
||
|
||
# 协程、任务和事件循环：和 JS Promise 对照
|
||
|
||
## 解释说明
|
||
|
||
| JavaScript | Python asyncio | 区别 |
|
||
| --- | --- | --- |
|
||
| 事件循环由运行时自动提供 | `asyncio.run(main())` 显式启动 | Python 脚本默认是同步的 |
|
||
| `async function f()` | `async def f():` | |
|
||
| 调用 async 函数**立即开始执行**，返回 Promise | 调用 `f()` 只得到**协程对象，不会执行** | 必须 `await` 或包装成任务 |
|
||
| `Promise.all([...])` | `await asyncio.gather(...)` 或 `TaskGroup` | |
|
||
| `Promise.race` | `asyncio.wait(..., return_when=FIRST_COMPLETED)` | |
|
||
| `setTimeout` 包装的延时 | `await asyncio.sleep(1)` | **不要用 `time.sleep`**，会阻塞整个循环 |
|
||
| 后台执行不等待 | `asyncio.create_task(coro)` | 要保存任务引用，否则可能被垃圾回收 |
|
||
| `for await (const x of stream)` | `async for x in stream:` | 流式读取模型输出常用 |
|
||
| `AbortController` | `task.cancel()`、`asyncio.timeout()` | |
|
||
|
||
**核心概念**
|
||
|
||
- **协程（coroutine）**：`async def` 定义的函数调用后得到的对象，在 `await` 处可以暂停，把控制权交还给事件循环。
|
||
- **任务（Task）**：被事件循环调度执行的协程。`create_task` 之后，它会在当前协程下一次 `await` 时开始运行。
|
||
- **事件循环**：单线程，不断地挑选“已经可以继续执行的任务”来运行。所以**任何一段不 await 的长时间计算或阻塞调用，都会让所有任务停下来**。
|
||
|
||
**适用场景**：IO 密集（网络请求、数据库、文件）。CPU 密集计算不会因为 asyncio 变快。
|
||
|
||
## 案例
|
||
|
||
RAG 接口需要同时做三件事：向量检索、关键词检索、查询用户权限。顺序执行约 300ms + 200ms + 100ms = 600ms；用 `gather` 并发执行，总耗时约等于最慢的那个，300ms。
|
||
|
||
## 代码示例
|
||
|
||
```python
|
||
import asyncio
|
||
import time
|
||
|
||
|
||
async def vector_search(q: str) -> list[str]:
|
||
await asyncio.sleep(0.3) # 模拟网络 IO
|
||
return [f"向量结果：{q}"]
|
||
|
||
|
||
async def keyword_search(q: str) -> list[str]:
|
||
await asyncio.sleep(0.2)
|
||
return [f"关键词结果：{q}"]
|
||
|
||
|
||
async def user_scopes(user_id: int) -> set[str]:
|
||
await asyncio.sleep(0.1)
|
||
return {"hr", "public"}
|
||
|
||
|
||
async def main() -> None:
|
||
# 1. 顺序执行：约 0.6 秒
|
||
start = time.perf_counter()
|
||
await vector_search("年假")
|
||
await keyword_search("年假")
|
||
await user_scopes(1)
|
||
print(f"顺序：{time.perf_counter() - start:.2f}s")
|
||
|
||
# 2. 并发执行：约 0.3 秒；结果顺序和参数顺序一致
|
||
start = time.perf_counter()
|
||
vec, kw, scopes = await asyncio.gather(vector_search("年假"), keyword_search("年假"), user_scopes(1))
|
||
print(f"并发：{time.perf_counter() - start:.2f}s", vec, kw, scopes)
|
||
|
||
# 3. 只调用不 await：协程不会执行，并且会有 "coroutine was never awaited" 警告
|
||
coro = vector_search("x")
|
||
print(type(coro).__name__) # coroutine
|
||
coro.close() # 演示用：手动关闭，避免警告
|
||
|
||
# 4. 后台任务：先启动，做别的事，再取结果
|
||
task = asyncio.create_task(vector_search("后台"))
|
||
await asyncio.sleep(0) # 让出一次控制权，任务开始执行
|
||
print("先做别的事")
|
||
print(await task)
|
||
|
||
# 5. 异步迭代：模拟模型流式输出
|
||
async def stream_tokens(text: str):
|
||
for ch in text:
|
||
await asyncio.sleep(0.01)
|
||
yield ch
|
||
|
||
print("".join([t async for t in stream_tokens("流式输出")]))
|
||
|
||
|
||
asyncio.run(main())
|
||
```
|
||
|
||
## 易错点
|
||
|
||
- **忘记 `await`**：拿到的是协程对象，函数根本没执行，常见表现是“结果是 `<coroutine object ...>`”。
|
||
- **在 async 函数里用 `time.sleep`、`requests.get`、同步数据库驱动**：整个服务卡住。用 `asyncio.sleep`、`httpx.AsyncClient`、异步驱动，或者 `asyncio.to_thread` 包装。
|
||
- **`create_task` 后不保存引用**：任务可能被垃圾回收而中途消失；也看不到它抛出的异常。
|
||
- **在已经运行的事件循环里再调用 `asyncio.run`**（比如在 Jupyter、FastAPI 的 async 路由里）：会报错，直接 `await` 即可。
|
||
- **以为 asyncio 是多线程**：它是单线程的协作式调度，共享变量在两个 `await` 之间是安全的，但跨 `await` 仍可能被别的任务修改。
|
||
|
||
## 面试常问
|
||
|
||
见 [asyncio 面试题](interview.md)：协程和线程的区别、事件循环原理、asyncio 适合什么场景、为什么不能在协程里调用阻塞函数。
|