python异步编程

asyncio 是 Python 3.4 版本引入的标准库,用于编写单线程并发代码,主要处理网络 I/O、文件 I/O 等。它使用 async / await 语法,让异步代码看起来像同步代码一样直观。

Python 异步编程主要就是:当一个任务正在等待网络、数据库、文件或定时器时,不让程序傻等,而是先去执行其他任务。 它解决的核心问题是:大量任务都要等待时,怎么更高效地利用这段等待时间。

1. 它解决的是什么问题

先用一个最直观的对比。

假设有三个接口要调,每个都要 2 秒:

1
2
3
4
5
6
7
8
9
import asyncio
import time


async def request_api(name: str) -> str:
print(f"{name} 开始")
await asyncio.sleep(2) # 模拟网络请求
print(f"{name} 完成")
return name

串行写法,一个一个来:

1
2
3
4
5
6
7
8
async def main() -> None:
a = await request_api("接口 A")
b = await request_api("接口 B")
c = await request_api("接口 C")
print([a, b, c])

asyncio.run(main())
# 耗时约 6 秒

并发写法,三个一起发出去:

1
2
3
4
5
6
7
8
9
10
async def main() -> None:
results = await asyncio.gather(
request_api("接口 A"),
request_api("接口 B"),
request_api("接口 C"),
)
print(results)

asyncio.run(main())
# 耗时约 2 秒

两段代码的差别只有一个词:第一段是 await 一个一个地等,第二段是把三个一起交给 asyncio.gather 去等。总耗时从”各人等各人的”变成”等最慢的那个”。

这就是 asyncio 的全部价值所在。它没有让单个请求变快,它让等待的时间被重叠起来。

2. 四个概念

概念 是什么
协程 Coroutine 用 async def 定义的异步函数。调用它不会立即执行,而是返回一个协程对象
事件循环 Event Loop 异步编程的核心,负责调度协程,在 I/O 等待时切换任务
Future 代表一个异步操作的最终结果,可以把它当作一个”未来完成”的容器
Task 是 Future 的子类,用于在事件循环中调度协程

三者的关系,按”谁管谁”排:

1
2
3
4
5
6
7
8
事件循环 Event Loop
负责调度和运行所有 Task
│
▼
Task 已经被安排运行的协程
│
▼
协程 Coroutine 要做的异步工作流程

用一句话概括:协程是”要做的事”,Task 是”已经排上号的那个事”,事件循环是”负责派活的调度员”。

还有一个容易被忽略的前提:asyncio 通常也运行在主线程上。 它不创建线程,只是在同一个线程里,趁某个协程在等 I/O 的时候切去跑另一个协程。

3. async / await 基础

async def 定义的是协程函数。它和普通函数最大的区别是:

1
2
3
4
async def fetch_data(url, delay):
print(f"开始获取 {url}")
await asyncio.sleep(delay)
return f"数据来自 {url}"

调用它不会立刻执行任何东西,只是返回一个协程对象:

1
2
coro = fetch_data("https://api1.com", 1)
print(coro) # <coroutine object fetch_data at 0x...>

要让它真的跑起来,得把它交给事件循环——用 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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def main():
# 创建多个任务
tasks = [
asyncio.create_task(fetch_data("https://api1.com", 1)),
asyncio.create_task(fetch_data("https://api2.com", 2)),
asyncio.create_task(fetch_data("https://api3.com", 1.5)),
]

# 等待所有任务完成
results = await asyncio.gather(*tasks)

for result in results:
print(result)

asyncio.run(main())

输出:

1
2
3
4
5
6
开始获取 https://api1.com
开始获取 https://api2.com
开始获取 https://api3.com
数据来自 https://api1.com
数据来自 https://api3.com
数据来自 https://api2.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
2
3
4
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
if isinstance(r, Exception):
print("这条失败了:", r)

5. create_task 之后,为什么不用立刻 await

很多人第一次看这段代码会觉得奇怪:

1
2
3
4
task1 = asyncio.create_task(background_task("1"))
task2 = asyncio.create_task(background_task("2"))

print("任务已创建")

任务明明没被 await,怎么就跑了?

因为 create_task() 会把任务添加到事件循环中进行调度,事件循环会自动运行这些任务,不需要显式 await。

具体发生了三件事:

  1. 任务被调度 —— 任务被添加到事件循环(event loop)中排队
  2. 立即执行 —— 任务会在事件循环的下一个周期开始执行
  3. 并发运行 —— 任务与当前代码以及其他任务并发运行

那 await 是干什么的?它的作用是:

  1. 等待任务完成 —— 暂停当前协程,直到任务完成
  2. 获取结果 —— 拿到任务的返回值
  3. 异常处理 —— 传播任务中发生的异常

但 await 不是触发任务执行的必要条件。

看一个能说明问题的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import asyncio

async def background_task(name):
print(f"任务 {name} 开始")
await asyncio.sleep(2)
print(f"任务 {name} 完成")
return f"结果 {name}"

async def main():
# 创建任务但不 await —— 任务会自动运行
task1 = asyncio.create_task(background_task("1"))
task2 = asyncio.create_task(background_task("2"))

print("任务已创建")

# 做一些其他工作
await asyncio.sleep(1)
print("主协程继续执行")

# 等待任务完成并获取结果
result1 = await task1
result2 = await task2

print(f"结果: {result1}, {result2}")

asyncio.run(main())

运行结果:

1
2
3
4
5
6
7
任务 1 开始
任务 2 开始
任务已创建
主协程继续执行
任务 1 完成
任务 2 完成
结果: 结果 1, 结果 2

两个”开始”排在”任务已创建”前面——说明 create_task 之后,两个协程立刻就跑到了它们各自的第一个 await,然后才轮到主协程继续打印。

所以在最外层,await task 和 create_task 是两件事: 前者是”要不要现在等结果”,后者是”有没有把这个协程排进调度”。当然,如果你创建了 Task 却从头到尾不 await,任务里抛的异常就没人接,Python 会打一条 Task exception was never retrieved 的警告——不 await 可以,但要保证异常有别的出口。

gather 那一节其实就属于”别的出口”:它一次接管了所有任务的等待和异常传播。

6. 取消和超时

task.cancel()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import asyncio

async def long_running_task():
try:
for i in range(10):
print(f"任务执行中... {i}")
await asyncio.sleep(1)
return "任务完成"
except asyncio.CancelledError:
print("任务被取消")
raise

async def main():
task = asyncio.create_task(long_running_task())

# 3 秒后取消任务
await asyncio.sleep(3)
task.cancel()

try:
await task
except asyncio.CancelledError:
print("捕获到取消异常")

asyncio.run(main())

三点要注意:

  • 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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import asyncio

async def slow_operation():
await asyncio.sleep(5)
return "完成"

async def main():
try:
task = asyncio.create_task(slow_operation())
# 设置 3 秒超时
result = await asyncio.wait_for(task, timeout=3.0)
print(result)
except asyncio.TimeoutError:
print("操作超时")
task.cancel()

asyncio.run(main())

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
2
async with asyncio.timeout(3.0):
await slow_operation()

7. 不要阻塞事件循环:asyncio.to_thread

这一节是整篇最容易踩坑的地方。

asyncio 是单线程的,所以协程里只要有一句同步阻塞代码,整个事件循环就卡住了——这一条卡住的不只是当前协程,是同一个循环里的所有协程。

把 await asyncio.sleep(2) 换成 time.sleep(2):

1
2
3
4
import time

async def bad():
time.sleep(2) # ❌ 整个事件循环停在这里 2 秒

asyncio.sleep() 是”把执行权交出去,2 秒后回来”;time.sleep() 是”占着线程睡 2 秒”。后者期间没有任何协程能动。

现实中的阻塞通常不长这样,而是调了一个同步库。LLM 的 Python SDK 大多是同步的,requests 也是同步的——直接这么写就会卡:

1
2
async def call_llm(self, **kwargs):
return self.client.chat.completions.create(**kwargs) # ❌ 同步调用,卡住整个循环

解法是 asyncio.to_thread()(Python 3.9+),把这次同步调用扔到一个线程里去执行:

1
2
3
4
response = await asyncio.to_thread(
self.client.chat.completions.create,
**kwargs
)

await 在等的时候,事件循环就空出来了,可以继续跑别的协程。

记住这条判断标准:协程里出现没有 await 的耗时调用,先问一句”它是同步的吗”。 是同步的就得包一层 to_thread,或者换一个原生支持 async 的库(httpx.AsyncClient、asyncpg、redis.asyncio 这些)。

顺带补一句上一篇文章里那个 with:

1
2
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(url, headers=headers, json=payload)

异步版本是 async with + await。对象管理的道理完全一样——退出代码块时自动 aclose(),return 也绕不过去。

8. Future:将来才有结果的占位对象

asyncio 里的 Future 可以理解成:一个将来才会有结果的占位对象。

它表示某个异步操作目前还没完成,但以后会得到:正常结果、异常,或者被取消。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import asyncio


async def set_result_later(future):
await asyncio.sleep(2)
future.set_result("任务完成")


async def main():
loop = asyncio.get_running_loop()

future = loop.create_future()

asyncio.create_task(set_result_later(future))

print("等待结果")

result = await future

print(result)


asyncio.run(main())

把过程拆开:

1
2
3
4
5
6
7
8
9
10
11
创建 Future,此时还没有结果
↓
创建后台任务
↓
await future,当前协程暂停
↓
两秒后调用 future.set_result()
↓
Future 完成
↓
main 恢复执行,得到结果

注意 loop.create_future() 和 asyncio.get_running_loop()——建 Future 要先拿到当前循环,因为 Future 是”挂在某个循环上”的。别用 asyncio.Future() 直接裸建。

不过说到底,平时写业务代码基本不需要自己建 Future。create_task 返回的就是一个 Future,gather 拿到的也是。Future 在这里的价值主要是理解概念:它解释了为什么 await 能”先暂停、以后再回来”——因为等的是一个可以被别人填结果的容器。

9. asyncio.Queue:协程之间传数据

asyncio.Queue 是专门给协程之间传递数据用的异步队列。

你可以把它理解成一个”任务传送带”:

1
2
3
4
5
生产者协程
↓
asyncio.Queue
↓
消费者协程

比如一个协程不断产生任务,另一个协程不断处理任务。

放和取

1
2
await queue.put(data)          # 往队列里放
task = await queue.get() # 从队列里拿

默认是 FIFO,先进先出:

1
2
3
队列里有:任务A  任务B  任务C
第一次 get → 任务A
剩下: 任务B 任务C

为什么 get() 前面要 await

这是 asyncio.Queue 最关键的地方。

假设队列现在是空的:

1
item = await queue.get()

消费者不会一直卡死主线程,而是:当前消费者协程暂停,把执行权交给事件循环,等队列有数据了再继续。

1
2
3
4
5
6
7
8
9
10
11
12
13
消费者发现 Queue 是空的
↓
await queue.get()
↓
消费者暂停
↓
事件循环去执行生产者
↓
生产者 queue.put("A")
↓
消费者被唤醒
↓
拿到 "A"

这就是它为什么适合 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
2
queue.empty()          # 现在是不是空的:True / False
queue.put_nowait(data) # 不等,直接塞。队列满了会抛 QueueFull

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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
ChiefArchitect / DeepScout / Writer ...
│
│ self.add_message(...)
▼
BaseAgent.add_message()
│
│ put_nowait(message)
▼
asyncio.Queue
│
│ await queue.get()
▼
run_agent_with_streaming()
│
│ yield msg
▼
SSE 前端

各个角色分工很清楚:

  • 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
2
3
4
self.add_message(state, "thought", {
"agent": self.name,
"content": "正在分析研究问题..."
})

这条消息会马上进入 Queue,然后外层:

1
2
msg = await message_queue.get()
yield msg

就能立刻推给前端,而不用等整个 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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def run_agent_with_streaming(agent, state):
task = asyncio.create_task(agent.run(state))
queue = state["_message_queue"]

# 只要 Agent 还没结束,就继续从 Queue 里读实时消息
while not task.done():
try:
msg = await asyncio.wait_for(queue.get(), timeout=0.5)
yield msg
except TimeoutError:
continue # 这半秒没消息,回去看一眼 task 是不是结束了

# Agent 已经结束,把 Queue 里剩下的消息全部取出来
while not queue.empty():
try:
yield queue.get_nowait()
except asyncio.QueueEmpty:
break

两个 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
2
3
进程   = 多个独立的公司          各自有自己的资源,互相不影响
线程 = 同一个公司里的多个员工 共享公司资源,动同一份数据要协调
协程 = 一个员工同时处理多个任务 谁在等,就先去做别的

选型上:

你的任务 用什么
大量 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 能让所有协程停摆。

参考

作者

wangyawei

发布于

2026-09-14

更新于

2026-09-14

许可协议