---
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 适合什么场景、为什么不能在协程里调用阻塞函数。