zhuangang/topics/python/asyncio/01-coroutines-tasks.md

id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week, order

id	title	type	domain	stack	tags	level	prerequisites	est_hours	related	week	order
python.asyncio.coroutines-tasks	协程、任务和事件循环：和 JS Promise 对照	lesson	python	asyncio	协程 事件循环 Task await async for	basic	python.python.syntax-for-js	2	python.asyncio.concurrency-control python.asyncio.interview	4	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。

代码示例

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 面试题：协程和线程的区别、事件循环原理、asyncio 适合什么场景、为什么不能在协程里调用阻塞函数。