0. 前言
手动建立许多个线程再利用回调函数构造调用链的时代已经结束了。
他们这么说。
协程体系通常被人们称为“并发编程的最终形态”。它完全避免了手动处理多个线程和复杂的回调函数嵌套，封装了一切关于回调的底层细节，使得我们能够用同步函数的思维来撰写异步函数。
本文旨在快速打通你的Python协程编程思维，以一种最贴合直觉的方式建立程序员对async/await语法的理解，服务于大家的Vibe Coding。
我们出发吧。
01. 概念快速入门
异步函数 | Async Func
1
2
3
async def func(spam: str)
result = await another_async_func(spam)
return result
在Python中，所谓异步函数是使用 async def 定义的函数。它的某些性质和使用方式与一般def的同步函数不同，使之可以在进行耗时等待操作时挂起自身并把CPU线程转移至其它函数和语句。也就是说，程序员可以通过一些特殊的写法（非常关键的await关键字的各种排布），使得原本单箭头的执行顺序分叉为两个甚至多个箭头，并控制其延伸的走向和合并的次序。
使用圆括号把实参表写在函数名右方进行调用时，与直接返回值的同步函数不同，异步函数返回协程对象而非立即执行。需要通过非常关键的await关键字来获取其返回值。
1
2
3
func: Callable[[str], Coroutine] # 异步函数本身
coro: coroutine = func(spam = "spam") # 传参后返回协程对象
result: str = await coro # await协程对象来获取结果
至于为什么要搞出这么一堆奇妙的语法，特别是非常关键的await关键字，下文会详细说明。
协程对象 | Coro
如前文所说，协程对象是异步函数进行圆括号传参后的返回值。Coro在建立后不会自动开始异步执行异步函数内的逻辑。我们一般把Coro传入asyncio.create_task()等函数之中进行管理，也可以直接使用await来执行该Coro内部的逻辑并获取返回值。
Coro只是一个概念，在一般情况下不需要被变量显式持有。result = await asyncFunc(raw)和task = asyncio.create_task(asyncFunc(raw)); result = await task足够解决大多数情况。
事件循环 | Event Loop
Python事件循环是异步编程的底层调度引擎，本质是一个在单线程内无限运行的协调调度器。它会在异步函数运行到await时挂起当前函数，并将执行路线切换到其它函数中。
从底层实现的角度看，事件循环维护着任务队列与就绪的集合，在每一次循环时检查并执行可运行的协程或回调。它通过多路复用等底层机制监控文件描述符、套接字和计时器，当协程运行到 await 挂起时，循环立即切换执行其他就绪任务，待 IO 事件或超时发生后恢复暂停的协程。
它确保了单线程内的并发性，驱动着 asyncio 生态的运行。
任务对象 | Task
Task是将Coro传入asyncio.create_task()等函数后的返回值。一旦Task被创建，其内部异步函数逻辑会立刻被事件循环进行评估并开始执行。之后，事件循环将自动识别该函数内部的await点来挂起Task把线程重新分配到创建Task的主分支。之后，与await Coro相似，主分支可以通过await Task来获取任务对象内部异步函数的返回值。
示例见下。
02. 干货在此 | 并发编程的常见模式
Fire-And-Forget | 聊聊何为分支
Fire-And-Forget 是最简单的异步编程模型之一，常用于日志记录、错误输出等不希望阻塞主线程、也不关心返回值的场景。我将在此场景中展示如何编写Async Func、用asyncio.create_task()建立分支，以及await Task来确保分支完成。
设想以下场景：
1
2
3
4
5
6
7
* raw = get_data()
|
* log(raw)
|
* result = process(data)
|
* end
上图展示了以同步形式调用log()的调度流程图。其缺点在于，如果log()是一个IO耗时函数（如文件、数据库的读写），后方process(data)会被直接阻塞，直到log()执行完毕。
1
2
3
4
5
6
7
* data = get_data()
| \
| * log(data)
|
* processed_data = process_data(data)
|
* end
上图展示了以异步形式调用log()的调度流程图。图中可见，在得到raw后，程序执行逻辑分为了两个不同的分支，主分支执行运算逻辑，侧分支执行日志记录。
我们尝试用Python实现这三个函数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def get_data():
with open('data.txt', mode='r') as f:
data = f.read()
return data
def process_data(data):
time.sleep(1)
return f"processed {data}"
async def log(data):
f = await aiofiles.open('app.log', mode='a')
await f.write(f"logged: {data}\n")
# 出于教学（偷懒？）目的，我没有使用async with语句，也没有使用try-finally
await f.close()
请看注释。请不要在任何地方这么写文件读写操作。如果你的Agent这么写，纠正它。
主分支的两个函数可以写成同步函数，而侧分支的log()写成了异步函数。async def说明调度器可以在内部任何一个await点（本例中有三个）将控制权切出该函数回到主分支。
三个await点所等待的三个函数由aiofiles提供，它们都是异步函数。异步函数传参后返回Coro， 而await Coro保证函数完成并返回该函数的返回值。
值得注意的是，await这一切出控制的特性反过来决定了使用await的函数必须是异步函数。同步函数不支持控制权的切出，因此，任何使用await的函数必须使用async def。试图在同步函数或程序顶层使用await会报SyntaxError语法错误。
习惯上，我们使用async def main()来做顶层封装，再交给asyncio.run(main())来执行。这多少弥补了Python没有main()的缺憾（雾）
1
2
3
4
5
6
async def main():
data = get_data() # 书写主分支
task = asyncio.create_task(log(data)) # 侧分支启动
processed_data = process_data(data) # 继续写主分支
print(processed_data)
await task
虽然我们不关心log()的返回值，但是我们需要确保它执行完毕。若不加上最后的await task，在主分支执行完毕时侧分支可能未能结束，此时该侧分支将会被中断。我们通过这种方法确保分支的合并，在end之前把侧分支并回主分支。
1
2
3
4
5
6
7
* data = get_data()
| \
| * log(data)
| |
* processed_data = process_data(data)
| /
* end
Fork-Join | 从分支中取得值
Fork-Join 是能够展现全套async/await语法的最简单模式。很简单，相比起Fire-And-Forget，Fork-Join 多了从分支中取值这一步。
我们先来让log返回值。
1
2
3
4
5
6
7
8
async def log(data):
try:
async with aiofiles.open('app.log', mode='a') as f: # 字面意思
await f.write(f"logged: {data}\n")
return True
except Exception as e:
print(e)
return False
aiofiles.open异步上下文管理器语法async with，其语法与
1
2
3
4
5
6
7
async def main():
data = get_data() # 书写主分支
task = asyncio.create_task(log(data)) # 侧分支启动
processed_data = process_data(data) # 继续写主分支
print(processed_data)
is_ok = await task
print("ok" if is_ok else "err")
Fan-In-Fan-Out | 批量展开分支并取值
Fan-In-Fan-Out 是 Fork-Join 的进阶模式，适用于需要同时发起多个独立 IO 任务、等待所有任务完成后再汇总结果的场景。我将在此场景中展示如何用 asyncio.gather() 同时启动多个协程，以及如何从多个分支中收集返回值。
设想以下场景：
1
2
3
4
5
6
7
8
9
* raw = fetch_data()
|
* result_a = process_a(raw)
|
* result_b = process_b(raw)
|
* combined = combine(result_a, result_b)
|
* end
上图展示了以同步形式调用 process_a() 和 process_b() 的调度流程图。其缺点在于，如果 process_a() 和 process_b() 都是IO 耗时函数（如调用不同的外部 API），它们会串行执行，总耗时等于两者之和。
1
2
3
4
5
6
7
8
* raw = fetch_data()
| \ \
| * result_a = process_a(raw) * result_b = process_b(raw)
| | |
| / /
* combined = combine(result_a, result_b)
|
* end
上图展示了以异步形式调用 process_a() 和 process_b() 的调度流程图。图中可见，在得到 raw 后，程序执行逻辑分为了两个并行的分支，两个分支同时执行各自的处理逻辑。当两个分支都完成后，主分支将两个结果合并。
我们尝试用 Python 实现这三个函数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def fetch_data():
with open('data.txt', mode='r') as f:
data = f.read()
return data
async def process_a(data):
await asyncio.sleep(2) # 模拟调用外部 API A
return f"processed by A: {data}"
async def process_b(data):
await asyncio.sleep(3) # 模拟调用外部 API B
return f"processed by B: {data}"
def combine(result_a, result_b):
return f"{result_a} | {result_b}"
两个处理函数都写成了异步函数，因为它们的耗时操作（网络请求、数据库查询等）可以被 await 挂起，让出控制权给事件循环。asyncio.gather() 会同时启动所有传入的协程，并等待它们全部完成后，按传入顺序返回结果列表。
1
2
3
4
5
6
7
8
async def main():
data = fetch_data() # 获取原始数据
result_a, result_b = await asyncio.gather( # 同时启动两个分支
process_a(data),
process_b(data)
)
combined = combine(result_a, result_b) # 合并结果
print(combined)
asyncio.gather() 接收多个协程对象（或可等待对象），将它们全部注册到事件循环中并发执行。当所有协程都完成时，返回一个包含所有返回值的元组。若某个协程抛出异常，gather() 默认会立即抛出该异常，其他协程仍会继续执行但结果会被丢弃。
1
2
3
4
5
6
7
8
9
* data = fetch_data()
| \
| * result_a = process_a(data) * result_b = process_b(data)
| | |
| * await asyncio.gather() 等待所有分支完成
| /
* combined = combine(result_a, result_b)
|
* end
此图由deepseek-v4-flash生成
可以通过for语句或列表推导式批量生成Coro，并将Coro列表通过星号解包的方式传入asyncio.gather()来批量启动任务。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
async def main():
data = fetch_data() # 获取原始数据
# 使用列表推导式批量生成 Coro
coros = [process_a(data) for _ in range(5)] # 展开五个分支
# Coro 的内部逻辑未开始运行
# 使用星号解包将 Coro 列表传入 gather
results = await asyncio.gather(*coros) # 相当于同时await五个Coro
# Coro 的内部逻辑此时才开始运行
# 处理所有结果
for i, result in enumerate(results):
print(f"Task {i}: {result}")
也可以传入Task。其优点在于除了启动任务外，额外地留出了主分支的空间。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
async def main():
data = fetch_data() # 获取原始数据
# 使用列表推导式批量创建 Task
tasks = [asyncio.create_task(process_a(data)) for _ in range(5)]
# Task的内部逻辑立即开始运行
# *在这里放一些你希望在得到result之前执行的代码*
# 将 Task 列表传入 gather
results = await asyncio.gather(*tasks)
# 起到确保完成的作用
# 处理所有结果
for i, result in enumerate(results):
print(f"Task {i}: {result}")
为了展示传入Task的用途，摘录一段我自己工程的代码：
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
27
28
29
30
31
32
async def run(self, items: list):
self.self_check()
# 部署所有机器到已有流水线
tasks = []
for machine in self.machines:
tasks += [
asyncio.create_task(machine.run())
for _ in range(machine.num)
]
# 向输入队列中添加所有项目
for item in items:
self.assembly_lines[0].put_nowait(item)
# 等待所有流水线完成
for line in self.assembly_lines:
await line.join()
# 向机器任务发送停止侦听信号
for task in tasks:
task.cancel()
# 等待所有机器任务完成
await asyncio.gather(*tasks, return_exceptions=True)
# 从输出队列中获取所有结果
out = []
while not self.output_queue.empty():
out.append(await self.output_queue.get())
return out
所有分支的异步函数是侦听性质的。主分支需要做出一些调度操作，被分支上的异步函数侦听到，进而执行内部逻辑。
由此可见，由于前者能够在扇出时额外留出主分支的代码操作空间，且能对各个任务进行精细处理，asyncio.gather(*tasks)可以处理的情况是asyncio.gather(*coros)的超集。而后者的好处是代码较少，task由gather内部封装并管理。
在Coding时，你可以根据需求跟你的Agent提一嘴。
Wait-For-First | 筛选最快返回的分支
Wait-For-First 与 Fan-In-Fan-Out有一定的相似性，其区别在于：Wait-For-First里所有分支的返回值是等价的，只需拿到一个返回值就可以放弃之后的所有分支。
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
27
28
29
30
31
32
33
34
35
36
37
async def fetch_from_cache():
await asyncio.sleep(0.5) # 模拟缓存较慢
return "cache_data"
async def fetch_from_db():
await asyncio.sleep(0.1) # 模拟数据库较快
return "db_data"
async def fetch_from_api():
await asyncio.sleep(0.8) # 模拟 API 最慢
return "api_data"
async def main():
# 创建任务
tasks = [
asyncio.create_task(fetch_from_cache()),
asyncio.create_task(fetch_from_db()),
asyncio.create_task(fetch_from_api())
]
# 等待第一个完成
done, pending = await asyncio.wait(
tasks,
return_when=asyncio.FIRST_COMPLETED
)
# 取第一个完成的结果
first_result = done.pop().result()
print(f"首个完成的任务返回: {first_result}")
# 取消剩余任务（避免资源浪费）
for task in pending:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
asyncio.run(main())
此模式直接依赖于asyncio.wait()。与 gather() 不同，wait() 提供了更加精准的控制手段，允许你指定放弃其余分支（如果需要）并返回的时间点。如等待所有任务完成，等待第一个完成，或是等待第一个异常抛出。通过return_when 指定，支持三种模式：
asyncio.ALL_COMPLETED：等待所有任务完成（默认）asyncio.FIRST_COMPLETED：只要有任意一个任务完成就返回asyncio.FIRST_EXCEPTION：只要有任意一个任务抛出异常就返回（若没有异常则等同ALL_COMPLETED）
asyncio.wait() 返回两个Task集合：done（已完成的任务）和 pending（未完成的任务）。应当手动task.cancel()并pass可能出现的asyncio.CancelledError来避免资源浪费。
3. 总结 我们学到了什么
我们一路从最基础的概念走到了并发编程的常见模式。
从异步函数与协程对象的区分开始，我们理解了 async def 定义的是函数本身，而圆括号传参后得到的 Coro 只是一个等待执行的对象，需要 await 才能真正驱动它。
事件循环作为单线程内的调度引擎，在每一个 await 点挂起当前协程、切换执行其他就绪任务，从而实现了看似并发的效果。
任务对象则是对协程的进一步封装，一旦通过 asyncio.create_task() 创建，其内部逻辑便会立即被事件循环调度执行，让我们能够在主分支继续工作的同时让侧分支并行推进。
在模式层面，我们从最简单的 Fire-And-Forget 出发，理解了如何用 create_task 建立分支、用 await task 合并分支，即使不关心返回值也要确保分支完成。
接着是 Fork-Join，在 Fire-And-Forget 的基础上增加了从分支中取值的步骤，展示了 await 获取返回值的完整链路。
Fan-In-Fan-Out 则进一步扩展了分支的数量，通过 asyncio.gather() 批量启动多个协程并等待全部完成，配合列表推导式和星号解包，能够优雅地处理任意数量的并行任务。
最后有 Wait-For-First，它展示了 asyncio.wait() 的灵活控制能力，在多个等价分支中去获取最快返回的结果，并主动取消剩余任务以避免资源浪费。
