python异步编程
asyncio 是 Python 3.4 版本引入的标准库,用于编写单线程并发代码,主要处理网络 I/O、文件 I/O 等。它使用 async / await 语法,让异步代码看起来像同步代码一样直观。
Python 异步编程主要就是:当一个任务正在等待网络、数据库、文件或定时器时,不让程序傻等,而是先去执行其他任务。 它解决的核心问题是:大量任务都要等待时,怎么更高效地利用这段等待时间。
1. 它解决的是什么问题
先用一个最直观的对比。
假设有三个接口要调,每个都要 2 秒:
1 | import asyncio |
串行写法,一个一个来:
1 | async def main() -> None: |
并发写法,三个一起发出去:
1 | async def main() -> None: |
两段代码的差别只有一个词:第一段是 await 一个一个地等,第二段是把三个一起交给 asyncio.gather 去等。总耗时从”各人等各人的”变成”等最慢的那个”。
这就是 asyncio 的全部价值所在。它没有让单个请求变快,它让等待的时间被重叠起来。
2. 四个概念
| 概念 | 是什么 |
|---|---|
| 协程 Coroutine | 用 async def 定义的异步函数。调用它不会立即执行,而是返回一个协程对象 |
| 事件循环 Event Loop | 异步编程的核心,负责调度协程,在 I/O 等待时切换任务 |
| Future | 代表一个异步操作的最终结果,可以把它当作一个”未来完成”的容器 |
| Task | 是 Future 的子类,用于在事件循环中调度协程 |
三者的关系,按”谁管谁”排:
1 | 事件循环 Event Loop |
用一句话概括:协程是”要做的事”,Task 是”已经排上号的那个事”,事件循环是”负责派活的调度员”。
还有一个容易被忽略的前提:asyncio 通常也运行在主线程上。 它不创建线程,只是在同一个线程里,趁某个协程在等 I/O 的时候切去跑另一个协程。
3. async / await 基础
async def 定义的是协程函数。它和普通函数最大的区别是:
1 | async def fetch_data(url, delay): |
调用它不会立刻执行任何东西,只是返回一个协程对象:
1 | coro = fetch_data("https://api1.com", 1) |
要让它真的跑起来,得把它交给事件循环——用 await,或者包成 Task:
1 | asyncio.run(fetch_data("https://api1.com", 1)) |
asyncio.run() 是程序入口:创建一个新的事件循环、跑完传进去的协程、关掉事件循环。
await 写在协程里面,含义是”在这里等这个操作完成,等的时候把执行权让出去“。这句话的后半截是关键——await 不只是”等”,它是”等,但不占着位置等”。
一个容易漏掉的规则:await 只能写在 async def 里面。 普通函数里写 await 直接语法错误。
4. 并发:create_task 和 gather
asyncio.create_task() 是 Python 3.7+ 引入的函数,用于创建一个 Task 对象。Task 是对协程的封装,可以让协程在事件循环中并发执行。
1 | async def main(): |
输出:
1 | 开始获取 https://api1.com |
注意三行”开始获取”是连着出现的——三个请求在第一个 await 处就全都发出去了,然后才按各自的耗时返回。总耗时由最慢的 2 秒那条决定,不是 1 + 2 + 1.5。
gather 的返回顺序和传入顺序一致,跟谁先完成无关。想按完成顺序拿,那是 asyncio.as_completed。
一个必须记住的例外
gather 默认 return_exceptions=False,意思是”第一个抛出来的异常会立刻往外传”。 这时候其它几个任务不会被取消,它们还在后台继续跑。
官方文档的原话是:其它 awaitable won’t be cancelled and will continue to run。
所以会出现这种情况:一个接口挂了,gather 当场抛异常,上层 except 之后函数就返回了,但剩下两个任务还挂在事件循环上。如果上层紧接着 asyncio.run() 结束,事件循环一关,它们就被直接掐掉——看起来像是”被取消了”,其实不是 gather 干的。
如果不希望一个失败影响整体,用 return_exceptions=True,异常会当成结果收进列表里:
1 | results = await asyncio.gather(*tasks, return_exceptions=True) |
5. create_task 之后,为什么不用立刻 await
很多人第一次看这段代码会觉得奇怪:
1 | task1 = asyncio.create_task(background_task("1")) |
任务明明没被 await,怎么就跑了?
因为 create_task() 会把任务添加到事件循环中进行调度,事件循环会自动运行这些任务,不需要显式 await。
具体发生了三件事:
- 任务被调度 —— 任务被添加到事件循环(event loop)中排队
- 立即执行 —— 任务会在事件循环的下一个周期开始执行
- 并发运行 —— 任务与当前代码以及其他任务并发运行
那 await 是干什么的?它的作用是:
- 等待任务完成 —— 暂停当前协程,直到任务完成
- 获取结果 —— 拿到任务的返回值
- 异常处理 —— 传播任务中发生的异常
但 await 不是触发任务执行的必要条件。
看一个能说明问题的例子:
1 | import asyncio |
运行结果:
1 | 任务 1 开始 |
两个”开始”排在”任务已创建”前面——说明 create_task 之后,两个协程立刻就跑到了它们各自的第一个 await,然后才轮到主协程继续打印。
所以在最外层,await task 和 create_task 是两件事: 前者是”要不要现在等结果”,后者是”有没有把这个协程排进调度”。当然,如果你创建了 Task 却从头到尾不 await,任务里抛的异常就没人接,Python 会打一条 Task exception was never retrieved 的警告——不 await 可以,但要保证异常有别的出口。
gather 那一节其实就属于”别的出口”:它一次接管了所有任务的等待和异常传播。
6. 取消和超时
task.cancel()
1 | import asyncio |
三点要注意:
cancel()是”请求取消”,不是”立刻停”。 它给协程抛一个CancelledError,但只有在这个协程跑到下一个await时才会真正抛出。所以上面await asyncio.sleep(1)那个点才是取消生效的地方。CancelledError继承的是BaseException,不是Exception。 所以except Exception抓不到它——这是故意的,免得你顺手把取消给吞了。- 协程内部抓到
CancelledError之后应该raise出去。 上面那个raise就是这个意思:打印完日志要继续往外抛,让await task那边也收到。吞掉不抛,上层会以为任务正常结束了,但实际上是半途停下的。
还有一点容易漏:取消之后最好 await 一下这个 task。 上面把它放在 try 里 await,是为了让取消有时间真正落地,不然警告和状态都不确定。
asyncio.wait_for
1 | import asyncio |
asyncio.wait_for() 的意思是:等一个异步操作,但最多只等指定时间。
这里就是:等 task 完成,但最多等 3 秒。3 秒内完成就返回结果;超过 3 秒还没完成,就抛 asyncio.TimeoutError。
有一处可以简化:wait_for 超时的时候,它自己已经把传进去的那个任务取消掉了(文档原话是 “If a timeout occurs, it cancels the task and raises TimeoutError”)。所以上面 except 里那句 task.cancel() 是多余的,留着无害,但知道它多余能省掉一次误判——排查”任务被取消了”这类问题时,记得 wait_for 本身就是嫌疑人之一。
两个版本上的细节:
- Python 3.11 起,
asyncio.TimeoutError就是内置的TimeoutError(asyncio.TimeoutError is TimeoutError为True)。所以老代码里写except asyncio.TimeoutError,新代码直接写except TimeoutError就行,现在只是留个向后兼容的别名。 - 3.11 还加了
asyncio.timeout()上下文管理器,写法更顺,而且比wait_for快一点,因为它不需要额外再造一个任务:
1 | async with asyncio.timeout(3.0): |
7. 不要阻塞事件循环:asyncio.to_thread
这一节是整篇最容易踩坑的地方。
asyncio 是单线程的,所以协程里只要有一句同步阻塞代码,整个事件循环就卡住了——这一条卡住的不只是当前协程,是同一个循环里的所有协程。
把 await asyncio.sleep(2) 换成 time.sleep(2):
1 | import time |
asyncio.sleep() 是”把执行权交出去,2 秒后回来”;time.sleep() 是”占着线程睡 2 秒”。后者期间没有任何协程能动。
现实中的阻塞通常不长这样,而是调了一个同步库。LLM 的 Python SDK 大多是同步的,requests 也是同步的——直接这么写就会卡:
1 | async def call_llm(self, **kwargs): |
解法是 asyncio.to_thread()(Python 3.9+),把这次同步调用扔到一个线程里去执行:
1 | response = await asyncio.to_thread( |
await 在等的时候,事件循环就空出来了,可以继续跑别的协程。
记住这条判断标准:协程里出现没有 await 的耗时调用,先问一句”它是同步的吗”。 是同步的就得包一层 to_thread,或者换一个原生支持 async 的库(httpx.AsyncClient、asyncpg、redis.asyncio 这些)。
顺带补一句上一篇文章里那个 with:
1 | async with httpx.AsyncClient(timeout=30.0) as client: |
异步版本是 async with + await。对象管理的道理完全一样——退出代码块时自动 aclose(),return 也绕不过去。
8. Future:将来才有结果的占位对象
asyncio 里的 Future 可以理解成:一个将来才会有结果的占位对象。
它表示某个异步操作目前还没完成,但以后会得到:正常结果、异常,或者被取消。
1 | import asyncio |
把过程拆开:
1 | 创建 Future,此时还没有结果 |
注意 loop.create_future() 和 asyncio.get_running_loop()——建 Future 要先拿到当前循环,因为 Future 是”挂在某个循环上”的。别用 asyncio.Future() 直接裸建。
不过说到底,平时写业务代码基本不需要自己建 Future。create_task 返回的就是一个 Future,gather 拿到的也是。Future 在这里的价值主要是理解概念:它解释了为什么 await 能”先暂停、以后再回来”——因为等的是一个可以被别人填结果的容器。
9. asyncio.Queue:协程之间传数据
asyncio.Queue 是专门给协程之间传递数据用的异步队列。
你可以把它理解成一个”任务传送带”:
1 | 生产者协程 |
比如一个协程不断产生任务,另一个协程不断处理任务。
放和取
1 | await queue.put(data) # 往队列里放 |
默认是 FIFO,先进先出:
1 | 队列里有:任务A 任务B 任务C |
为什么 get() 前面要 await
这是 asyncio.Queue 最关键的地方。
假设队列现在是空的:
1 | item = await queue.get() |
消费者不会一直卡死主线程,而是:当前消费者协程暂停,把执行权交给事件循环,等队列有数据了再继续。
1 | 消费者发现 Queue 是空的 |
这就是它为什么适合 asyncio——等待本身也是异步的。 换成同步队列,消费者会真的占着线程堵在那儿;这里是”睡着了,等有人叫”。
三种取法
| 写法 | 队列为空时 |
|---|---|
await queue.get() |
一直异步等,直到有数据 |
await asyncio.wait_for(queue.get(), 0.5) |
最多等 0.5 秒,超时抛 TimeoutError(同时取消内部那个 get() 的等待) |
queue.get_nowait() |
不等。有就拿,没有立刻抛 asyncio.QueueEmpty |
用一句话记:
await queue.get()→ 没消息就一直等await asyncio.wait_for(queue.get(), 0.5)→ 没消息最多等 0.5 秒queue.get_nowait()→ 不等,有就拿,没有就直接报错
wait_for 那一档的用处是”别永远卡在 Queue 上“:等待期间可以醒过来,检查一下别的状态(比如生产者是不是已经结束了)。
另外两个
1 | queue.empty() # 现在是不是空的:True / False |
asyncio.Queue() 创建时如果不设 maxsize,默认可以一直放,所以 put_nowait 基本不会失败——这也是它常被拿来在同步方法里发消息的原因(见下一节)。
要做一个”等所有任务都处理完”的汇合点,用 join():
1 | await queue.join() |
但它有个必须配对的兄弟:每取走一条并处理完,要调用 queue.task_done()。 否则 join() 永远不返回——它数的是”put 进去多少条、task_done 了多少条”,少一次就卡死。这一条不知道的话,很容易写出一个静默挂住的程序。
10. 一个实战:Agent 和 SSE 之间的消息管道
这套东西在真实项目里的样子可以看一个具体场景:Agent 在后台跑,同时要把实时消息一条条推给前端。
asyncio.Queue 在这里相当于 Agent 和 SSE 输出之间的异步消息管道:
1 | ChiefArchitect / DeepScout / Writer ... |
各个角色分工很清楚:
asyncio.create_task()→ 让 Agent 后台运行asyncio.Queue()→ Agent 和工作流之间传消息await queue.get()→ 异步等待下一条消息asyncio.wait_for(..., 0.5)→ 避免永远卡在 Queue 上,可以周期性检查取消和完成状态queue.get_nowait()→ Agent 结束后清掉剩余消息async for+yield→ 最终形成 SSE 实时流
比如 ChiefArchitect 里:
1 | self.add_message(state, "thought", { |
这条消息会马上进入 Queue,然后外层:
1 | msg = await message_queue.get() |
就能立刻推给前端,而不用等整个 Agent 执行结束。
为什么这里用 put_nowait 而不是 await put
1 | state["_message_queue"].put_nowait(message) |
而不是:
1 | await state["_message_queue"].put(message) |
因为这个 Queue 创建时没有设置 maxsize,默认基本可以一直放,所以这里希望发消息不等待,直接塞进去。
好处直接体现在接口签名上:add_message() 可以保持普通同步方法:
1 | def add_message(...) # 不用改成 async def |
如果用了 await put,add_message 就得改成 async def,然后所有调它的地方都得跟着 await,一路传染上去。 这是同步 / 异步边界上一个很实际的取舍。
消费那一侧
把生产者和消费者拼起来,模式大致是这样:
1 | async def run_agent_with_streaming(agent, state): |
两个 while 各管一件事:
- 第一个循环:
task.done()是”检查这个异步任务是不是已经结束了”。没结束就从 Queue 里读;wait_for的 0.5 秒超时让循环每半秒醒一次,回去重新判断task.done()——否则 Agent 结束后一直没新消息,这个await queue.get()会永远挂着,SSE 流永远不结束。 - 第二个循环:Agent 结束后清空队列。
empty()返回True表示空了、False表示还有消息,所以while not queue.empty()就是”只要还有消息就继续取”。这里用get_nowait()而不是await get(),因为已经确定不再等了。外层再包一个except: break,是防empty()和get_nowait()之间出现竞态。
一个小提醒:while not task.done() 退出时没有 await task,所以如果 Agent 是抛异常结束的,那个异常就被丢掉了,Python 会打一条 Task exception was never retrieved。想让它冒出来,在循环之后补一次 await task。
一句话总结这个 Queue 的作用:Agent 用 add_message() 往 Queue 里生产实时事件,工作流从 Queue 里消费事件,再通过 SSE 推给前端。
11. 线程、进程、协程
最后把这三个放在一起。
| 进程 | 线程 | 协程 | |
|---|---|---|---|
| 调度者 | 操作系统 | 操作系统 | 事件循环(程序自己) |
| 资源 | 独立内存空间,互相隔离 | 共享进程内存 | 共享所在线程的内存 |
| 切换时机 | 随时,由 OS 决定 | 随时,由 OS 决定 | 只在 await 处主动让出 |
| 开销 | 最大 | 中等,每个线程一条独立栈 | 最小,一个线程能跑大量协程 |
| 能并行吗 | 能,多核并行 | 受 GIL 限制,纯 Python 字节码不能同时跑;I/O 可以并发 | 单线程并发,不是并行 |
| 阻塞的影响 | 只卡自己 | 只卡自己 | 卡住整个事件循环 |
| 适合 | CPU 密集 | 同步阻塞 I/O、老的同步 SDK | 大量异步 I/O |
几点展开:
调度者不同。 线程是操作系统调度的执行单元,创建之后由操作系统决定什么时候运行、暂停和切换;协程是用户态的轻量级执行单元,由程序主动调度,执行到 await 才暂停让出。线程切换是被动的,协程切换是主动的。
切换方式不同。 线程可能在任意时刻被 OS 切走;协程只在 await 处切。这带来一个直接后果:协程之间的代码在两次 await 之间是不被打断的,所以共享数据通常不用加锁——这也是为什么 asyncio 里很少见到 Lock。
GIL 那一栏要说清。 在普通 CPython 中,由于 GIL,多个线程通常不能同时执行纯 Python 字节码;但 I/O 操作会释放 GIL,所以 I/O 并发是有效的。这正好解释了为什么线程和协程都适合 I/O、都不适合 CPU 密集。
阻塞影响不同。 一个线程执行阻塞操作,通常只阻塞该线程。但协程里执行同步阻塞代码,会阻塞整个事件循环,同一循环里的其他协程全部停摆——这就是上一节 to_thread 存在的原因。
最容易误解的一条:threading.Thread 会创建一条真正的线程;asyncio.create_task() 不创建线程,只是在事件循环中安排一个协程执行。
用三个角色打比方:
1 | 进程 = 多个独立的公司 各自有自己的资源,互相不影响 |
选型上:
| 你的任务 | 用什么 |
|---|---|
| 大量 CPU 计算 | 多进程,这样才能吃满多个核心 |
| 同步阻塞 I/O、不支持的异步的老 SDK | 多线程 |
大量异步 I/O,且库支持 async / await |
协程 |
协程这类里常见的场景:调用大模型 API、HTTP 请求、异步数据库查询、Redis 请求、WebSocket、同时处理大量用户连接、同时调用多个 Agent 工具、FastAPI 中的异步接口。
一句话总结:进程隔离性最好但开销最大,线程共享资源但要注意线程安全,协程最轻量,适合高并发的 I/O 场景。
12. 小结
真正要记住的是四件事:
一、asyncio 的价值是重叠等待时间,不是让单个操作变快。 串行 await 三次和 gather 一次,耗时差三倍。
二、create_task 是”排上号”,await 是”等结果”。 任务不需要 await 也会跑,但异常得有出口,否则会得到 Task exception was never retrieved。
三、协程里不能有同步阻塞调用。 它卡的不是当前协程,是整个事件循环。同步 SDK 一律包 asyncio.to_thread,或者换成原生异步的库。
四、await 的位置就是切换点。 协程只在 await 处让出执行权——这既解释了为什么不用加锁,也解释了为什么一个 time.sleep 能让所有协程停摆。
参考
asyncio总览:https://docs.python.org/zh-cn/3/library/asyncio.html- 协程与任务(
create_task/gather/wait_for/to_thread):https://docs.python.org/zh-cn/3/library/asyncio-task.html - 队列(
asyncio.Queue):https://docs.python.org/zh-cn/3/library/asyncio-queue.html - 异步异常(
CancelledError/TimeoutError):https://docs.python.org/zh-cn/3/library/asyncio-exceptions.html - 同步原语与
Queue的同步版本:https://docs.python.org/zh-cn/3/library/asyncio-sync.html - 并发执行(
threading/multiprocessing/concurrent.futures):https://docs.python.org/zh-cn/3/library/concurrency.html contextlib.asynccontextmanager:https://docs.python.org/zh-cn/3/library/contextlib.html#contextlib.asynccontextmanager