【Python】并发——异步
·
目录
并发
异步
基本概念与原理
同步 vs 异步 vs 并发 vs 并行
这些名词看似相似，但一定要区分清楚：
| 概念 | 解释 | 示例 |
|---|---|---|
| 同步 (Synchronous) | 任务一个接一个执行，前一个没完成下一个不能开始。 | 打电话时等对方说完你才能说。 |
| 异步 (Asynchronous) | 发出任务后不等待结果，继续干别的事。 | 发短信，不用等对方回。 |
| 并发 (Concurrency) | 同一时间“看起来”有多个任务在执行（任务切换快）。 | 一个厨师做多道菜，来回切换。 |
| 并行 (Parallelism) | 真正的同时执行多个任务（多核CPU并行）。 | 多个厨师同时做不同的菜。 |
Python 异步属于 并发 的一种形式（单线程多任务），不是多核“并行”，但能极大提升 I/O 密集任务的性能。
同步阻塞的弊端
假设我们要同时下载 3 个网页：
import time
import requests
def download(url):
print("开始下载", url)
time.sleep(2)
print("下载完成", url)
urls = ['a.com', 'b.com', 'c.com']
start = time.time()
for u in urls:
download(u)
print("总耗时:", time.time() - start)
输出：
开始下载 a.com
下载完成 a.com
开始下载 b.com
下载完成 b.com
开始下载 c.com
下载完成 c.com
总耗时: 6.00s
这是同步阻塞：下载是串行的，一个任务没完成，下一个不能开始。
异步的核心思想
异步的本质是：
在等待I/O时不阻塞整个程序，而是切换去执行别的任务。
比如：
- 网络请求、文件I/O、数据库查询都属于“等待”；
- 这些时间 CPU 是空闲的；
- 异步就利用这些空闲时间去干别的活。
事件循环（Event Loop）的概念
异步编程的核心机制就是 事件循环：
- 注册任务 (coroutine)
- 启动事件循环 event loop
- 当任务等待I/O → 暂停
- I/O完成 → 恢复执行
也就是说：
- 事件循环负责调度协程的执行；
- 当一个协程遇到等待（I/O），事件循环就去执行别的协程；
- 这样单线程也能高效利用 CPU 时间。
Python 异步的演化历史
| 阶段 | 特点 |
|---|---|
| Python 3.3 | 引入 yield from（生成器异步） |
| Python 3.5 | 正式加入 async / await 语法 |
| Python 3.6+ | asyncio 模块成熟，标准化 |
| Python 3.7+ | 默认事件循环 API 更加简洁（asyncio.run） |
| Python 3.11+ | 异步性能大幅提升（TaskGroup、ExceptionGroup） |
异步的本质是协程（Coroutine）
你可以把协程理解成一种“能暂停和恢复执行”的函数。
普通函数执行到底就结束；
协程可以：
await等待一个异步操作；- 等待完成后恢复原地继续执行。
这样程序不会因为等待卡住，而是可以去处理别的任务。
同步 vs 异步
同步版本
import time
def func1():
time.sleep(2)
print("任务1完成")
def func2():
time.sleep(2)
print("任务2完成")
def main():
func1()
func2()
main()
耗时约 4 秒。
异步版本
import asyncio
async def func1():
await asyncio.sleep(2)
print("任务1完成")
async def func2():
await asyncio.sleep(2)
print("任务2完成")
async def main():
await asyncio.gather(func1(), func2())
asyncio.run(main())
耗时约 2 秒
原因：await asyncio.sleep(2) 时任务1挂起，事件循环立刻执行任务2。
async / await 语法
为什么要有 async / await
在 Python 3.5 之前，异步代码是通过生成器和 yield from 实现的，语法复杂、可读性差。
于是 Python 3.5 引入了 新的关键字：async 和 await，使协程代码的结构更直观。
async 的作用：定义协程函数
在普通函数前加上 async，就变成了一个协程函数（coroutine function）。
async def say_hello():
print("Hello")
- 调用普通函数：立即执行。
- 调用协程函数：不会立即执行，而是返回一个协程对象（coroutine object）。
举个例子：
async def test():
print("Hello Async")
result = test()
print(result)
输出：
<coroutine object test at 0x0000021F6...>
注意：
调用协程函数时，代码不会执行，必须由事件循环（event loop）调度执行。
如何执行协程函数
有三种主要方式执行协程：
asyncio.run()
推荐的最高层封装方式（Python 3.7+）：
输出：import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") asyncio.run(main())Hello Worldasyncio.run()会：- 自动创建事件循环；
- 执行协程；
- 等待所有任务结束；
- 关闭事件循环。
- 手动获取事件循环（低层 API）
import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") loop = asyncio.get_event_loop() loop.run_until_complete(main()) loop.close()这种方式更灵活，但一般只有在框架内部（如 FastAPI）才会使用。
- 在异步函数内部用
awaitasync def func1(): print("A") await asyncio.sleep(1) print("B") async def main(): await func1() # await 执行另一个协程 asyncio.run(main())await只能出现在 async 函数内部。
否则会报错：SyntaxError: 'await' outside async function
await 的作用await 的本质是：
暂停当前协程的执行，等待另一个协程完成，然后恢复执行。
比如：
import asyncio
async def task1():
print("开始任务1")
await asyncio.sleep(2)
print("任务1完成")
async def task2():
print("开始任务2")
await asyncio.sleep(1)
print("任务2完成")
async def main():
await task1()
await task2()
asyncio.run(main())
输出：
开始任务1
任务1完成
开始任务2
任务2完成
这是顺序执行，因为每次 await 都等待前一个协程结束。
多个协程并发执行（asyncio.gather）
要真正并发执行多个协程，可以用 asyncio.gather()：
import asyncio
async def task1():
print("开始任务1")
await asyncio.sleep(2)
print("任务1完成")
async def task2():
print("开始任务2")
await asyncio.sleep(1)
print("任务2完成")
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())
输出：
开始任务1
开始任务2
任务2完成
任务1完成
耗时约 2 秒，而不是 3 秒。
协程的返回值
协程可以像普通函数一样返回值：
import asyncio
async def add(x, y):
await asyncio.sleep(1)
return x + y
async def main():
result = await add(3, 4)
print(result)
asyncio.run(main())
输出：
7
协程嵌套调用
协程可以调用另一个协程：
async def inner():
print("Inner start")
await asyncio.sleep(1)
print("Inner end")
async def outer():
print("Outer start")
await inner()
print("Outer end")
asyncio.run(outer())
常见错误与陷阱
| 错误 | 原因 |
|---|---|
await 在 async 外使用 |
await 只能在 async def 中使用 |
忘记用 await 调用协程 |
协程不会自动执行，必须 await |
用 time.sleep() 阻塞事件循环 |
异步程序中应使用 await asyncio.sleep() |
| 在同步代码中调用协程 | 必须用 asyncio.run() 或事件循环执行 |
实战例子：异步下载任务
import asyncio
async def download(url):
print(f"开始下载: {url}")
await asyncio.sleep(2)
print(f"下载完成: {url}")
async def main():
urls = ['a.com', 'b.com', 'c.com']
tasks = [download(u) for u in urls]
await asyncio.gather(*tasks)
asyncio.run(main())
输出：
开始下载: a.com
开始下载: b.com
开始下载: c.com
下载完成: b.com
下载完成: a.com
下载完成: c.com
说明 3 个下载任务是并发执行的。
协程（Coroutine）的本质与生命周期
什么是协程（Coroutine）
概念定义
协程（Coroutine）是一种 比线程更轻量级的并发单元。
它在一个线程中就可以实现并发执行，通过在函数之间主动让出执行权实现“伪并行”。
一句话解释：
“协程是可以在执行过程中暂停、并在之后恢复执行的函数。”
协程 vs 线程
| 对比项 | 协程 (Coroutine) | 线程 (Thread) |
|---|---|---|
| 切换控制 | 程序员控制（显式 await） |
操作系统控制（时间片） |
| 上下文切换 | 在用户态完成，几乎无开销 | 需要进入内核态，成本高 |
| 并发模式 | 单线程内伪并发 | 多线程并行 |
| 同步机制 | 不需要锁（协作式） | 需要锁防止竞争 |
| 使用场景 | IO密集型 | CPU密集型 |
协程的历史
协程并不是 Python 独有的概念，最早源自 1960 年代 ALGOL 语言。
Python 从：
- Python 2.5 起支持
yield（生成器级协程雏形）； - Python 3.5 起引入
async/await原生语法； - Python 3.7+ 起正式稳定
asyncio模块。
协程的底层本质
Python 中的协程实际上是一种特殊的 生成器（generator）。
协程 = 生成器的高级形式
生成器可以暂停 (yield) 与恢复 (send)，协程在此基础上增加了：
- 状态管理（READY、RUNNING、SUSPENDED、FINISHED）
- 自动调度（由事件循环驱动）
- 语法糖：
async def、await
例如：
def gen_coroutine():
x = yield "start"
print("received:", x)
yield "end"
g = gen_coroutine()
print(next(g)) # 输出 start
print(g.send(42)) # 输出 received: 42
这已经展示了协程的“暂停→恢复”特性。
协程的生命周期（Lifecycle）
协程的生命周期由 asyncio 或事件循环管理，可以分为以下几个阶段：
| 阶段 | 状态名称 | 说明 |
|---|---|---|
| 1 | Created | 定义了 async def 函数，但未执行 |
| 2 | Scheduled | 协程对象交给事件循环调度 |
| 3 | Running | 正在执行，直到遇到 await 或返回 |
| 4 | Suspended | 暂停等待另一个协程/IO 完成 |
| 5 | Finished | 执行完毕或抛出异常 |
示例：观察协程的创建与运行
import asyncio
async def my_coroutine():
print("Start coroutine")
await asyncio.sleep(1)
print("End coroutine")
return 42
async def main():
coro = my_coroutine()
print("协程创建完成：", coro)
result = await coro
print("协程结果：", result)
asyncio.run(main())
输出：
协程创建完成： <coroutine object my_coroutine at 0x...>
Start coroutine
End coroutine
协程结果： 42
说明：
my_coroutine()创建了一个协程对象，但不会立即执行；- 只有当
await coro时，事件循环才调度它运行； - 协程运行中遇到
await asyncio.sleep(1)时会挂起，让出控制权； - 最后恢复执行并返回结果。
协程的暂停与恢复
核心关键字：
await：挂起当前协程，等待另一个协程完成；async：定义异步函数，使其返回协程对象。
示例：
import asyncio
async def task(name, delay):
print(f"{name} 开始")
await asyncio.sleep(delay)
print(f"{name} 结束")
return f"{name} 完成"
async def main():
res1 = await task("任务1", 2)
res2 = await task("任务2", 1)
print(res1, res2)
asyncio.run(main())
输出：
任务1 开始
任务1 结束
任务2 开始
任务2 结束
任务1 完成 任务2 完成
说明：await 会等待前一个任务完成，所以上面是“顺序执行”。
协程的调度机制（Event Loop 简介）
协程的运行依靠 事件循环（Event Loop），它是异步的“心脏”。
事件循环的核心职责：
- 管理任务队列；
- 调度协程运行；
- 在 I/O 完成时唤醒暂停的协程。
流程示意：
创建协程对象 → 事件循环调度 → 执行直到 await → 挂起 → IO完成唤醒 → 继续执行
asyncio 事件循环（Event Loop）原理
事件循环是什么？
事件循环（Event Loop） 是 Python 异步编程的“心脏”。
一句话解释：
事件循环负责管理所有协程的调度，决定哪个协程什么时候运行、什么时候暂停、什么时候恢复。
所有 async/await 异步任务，最终都由事件循环来统一管理。
类比理解
可以把事件循环想象成一个 “总调度员”：
- 它有一个“任务队列”；
- 每当有协程准备好执行，就把它放进队列；
- 遇到
await（例如await asyncio.sleep()）时，协程会主动让出执行权； - 当被等待的 I/O 事件完成后，事件循环会再次调度这个协程继续执行。
这是一种协作式调度（Cooperative Scheduling），
不同于多线程的“抢占式调度”。
事件循环的基本结构
import asyncio
async def main():
print("Hello ...")
await asyncio.sleep(1)
print("... World!")
asyncio.run(main())
运行流程：
- 创建事件循环（loop）
- 将
main()协程封装为任务（Task） - loop 启动，调度任务运行
main()执行到await asyncio.sleep(1)时挂起- loop 暂时执行其他任务
- 1 秒后 I/O 完成，loop 唤醒
main()，继续执行 main()结束后，loop 关闭
实际内部机制（asyncio.run 的过程）
asyncio.run(main()) 内部做了三件事：
def run(main):
loop = asyncio.new_event_loop() # 1. 创建事件循环
asyncio.set_event_loop(loop)
result = loop.run_until_complete(main) # 2. 执行主协程
loop.close() # 3. 关闭事件循环
return result
事件循环的关键方法与职责
事件循环对象（asyncio.AbstractEventLoop）有许多重要方法。
| 方法 | 作用 |
|---|---|
run_until_complete(task) |
运行任务直到完成 |
run_forever() |
一直运行循环（常用于服务器） |
create_task(coro) |
把协程包装成可调度的任务 |
call_soon(callback) |
尽快执行回调函数 |
call_later(delay, callback) |
延迟执行回调 |
stop() / close() |
停止或关闭事件循环 |
示例：手动控制事件循环
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1)
print("World")
# 手动管理事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
task = loop.create_task(say_hello())
loop.run_until_complete(task)
loop.close()
上面这段代码与 asyncio.run(say_hello()) 本质一样，只是我们手动控制了 loop 的生命周期。
事件循环的核心结构原理
事件循环的核心逻辑可以用伪代码描述如下：
while tasks_not_done:
ready_tasks = get_ready_tasks() # 获取可运行的协程
for task in ready_tasks:
try:
task.step() # 运行到下一个 await
except StopIteration:
mark_task_done(task)
handle_io_events() # 检查 I/O 是否完成
说明：
task.step()是协程运行到下一个await的过程；- 当协程挂起时，事件循环会记录“等待的I/O”；
- 一旦 I/O 完成，它会唤醒对应协程。
整个过程完全在单线程中完成，无锁、无竞争。
事件循环与 I/O 多路复用（Selector）
Python 的事件循环实际上是基于 I/O 多路复用机制（如 epoll、select、kqueue）。
简单来说：
操作系统能同时监控多个文件描述符（socket、pipe等），
当任意一个 I/O 完成时，通知事件循环继续执行相关协程。
示意图：
┌────────────────────────────┐
│ Event Loop │
├────────────────────────────┤
│ Tasks Queue │
│ Sleeping Tasks │
│ I/O Watcher (epoll/select) │
│ Callbacks Queue │
└────────────────────────────┘
所以说：
- 协程是语言层的并发机制；
- 事件循环是运行时的调度引擎；
- I/O 多路复用是操作系统层的底层支撑。
三者共同构成 Python 异步的核心体系。
事件循环的嵌套与多循环问题
Python 每个线程最多只能有一个事件循环。
常见错误：
RuntimeError: asyncio.run() cannot be called from a running event loop
原因是：
- Jupyter Notebook、Qt、某些框架中已经有一个事件循环在运行；
- 你又在其中调用了
asyncio.run()。
解决办法：
- 用
await直接运行； - 或使用
nest_asyncio（非标准库）。
事件循环的生命周期
| 阶段 | 说明 |
|---|---|
| 1. 创建（create） | loop = asyncio.new_event_loop() |
| 2. 绑定（set） | asyncio.set_event_loop(loop) |
| 3. 执行（run） | loop.run_until_complete(task) |
| 4. 关闭（close） | loop.close() |
Task 与 Future（任务与未来对象）
为什么要有 Task 和 Future？
前面了解到：
事件循环（Event Loop）负责调度协程运行。
但是——事件循环不能直接运行协程对象，它需要一种可追踪执行状态的“封装体”，能记录协程的：
- 是否运行中？
- 是否完成？
- 是否抛出异常？
- 最终结果是什么？
于是就有了：
Task：封装协程的执行单元；Future：表示“将来某个时间会有结果的对象”。
Future 是什么？（未来对象）
-
概念定义
Future 表示一个尚未完成的计算结果，将来某个时刻会被设置（set result）。
可以把
Future想象成一个“空盒子”：- 你先拿到盒子（Future 对象）；
- 稍后，计算完成时，盒子里才装入结果；
- 你可以“等待”盒子装满（即 await Future）。
-
创建与使用 Future
import asyncio async def main(): loop = asyncio.get_running_loop() future = loop.create_future() # 创建一个空Future # 模拟异步操作 async def set_result_later(): await asyncio.sleep(1) future.set_result("任务完成") # 安排任务执行 loop.create_task(set_result_later()) print("等待Future结果中...") result = await future print("结果:", result) asyncio.run(main())输出：
等待Future结果中... 结果: 任务完成说明：
Future是“结果容器”；await future会挂起，直到set_result()被调用；loop.create_task()则负责调度异步任务。
-
Future 的状态机
状态 含义 设置方式 Pending 未完成 默认状态 Running 正在执行 loop 调度中 Finished 正常完成 set_result(value)Cancelled 被取消 cancel()Exception 出现异常 set_exception(error) -
手动操作 Future
import asyncio async def main(): fut = asyncio.Future() print("初始状态:", fut.done()) fut.set_result("OK") print("是否完成:", fut.done()) print("结果:", fut.result()) asyncio.run(main())输出：
初始状态: False 是否完成: True 结果: OK
Task 是什么？（任务对象）
-
定义
Task 是
Future的一个子类，用来自动执行协程并保存结果。不需要手动设置结果，
Task 会：- 启动协程；
- 在协程结束时自动
set_result()。
-
创建 Task
import asyncio async def say_hello(): await asyncio.sleep(1) return "Hello Async World" async def main(): task = asyncio.create_task(say_hello()) # 创建任务 print("任务已调度:", task) result = await task print("任务结果:", result) asyncio.run(main())输出：
任务已调度: <Task pending name='Task-1' coro=<say_hello() running at ...>> 任务结果: Hello Async World -
Task 与 Future 的关系
对比项 Future Task 作用 表示“未来结果” 封装并运行一个协程 是否自动执行 否 会自动启动协程 设置结果 手动调用 set_result()协程完成后自动设置 常用场景 自定义异步回调、网络通信底层 异步任务调度 关系图：
协程（Coroutine） ↓ 被封装 任务（Task） → 继承 → Future ↓ 返回结果（通过 await 获取） -
批量创建与等待任务
import asyncio async def task(name, delay): await asyncio.sleep(delay) return f"{name} 完成" async def main(): tasks = [ asyncio.create_task(task("任务A", 2)), asyncio.create_task(task("任务B", 1)) ] results = await asyncio.gather(*tasks) print("所有结果:", results) asyncio.run(main())输出：
所有结果: ['任务A 完成', '任务B 完成']asyncio.gather()会并发运行所有任务，并在全部完成后返回结果列表。
Task 的生命周期
| 阶段 | 状态 | 说明 |
|---|---|---|
| 1 | Created | create_task() 创建后，尚未运行 |
| 2 | Scheduled | 被加入事件循环队列 |
| 3 | Running | 协程正在执行 |
| 4 | Pending | 遇到 await 被挂起 |
| 5 | Finished | 协程执行完毕，Task 自动完成 |
可以用 task.done()、task.cancelled()、task.result() 查看状态。
例子：状态追踪
import asyncio
async def demo():
await asyncio.sleep(1)
return "Done"
async def main():
t = asyncio.create_task(demo())
print("初始状态:", t.done())
await asyncio.sleep(0.5)
print("中途状态:", t.done())
await t
print("结束状态:", t.done())
print("结果:", t.result())
asyncio.run(main())
输出：
初始状态: False
中途状态: False
结束状态: True
结果: Done
Task 的取消与异常
import asyncio
async def never_end():
try:
while True:
print("工作中...")
await asyncio.sleep(0.5)
except asyncio.CancelledError:
print("任务被取消！")
async def main():
t = asyncio.create_task(never_end())
await asyncio.sleep(2)
t.cancel() # 取消任务
await t
asyncio.run(main())
输出：
工作中...
工作中...
工作中...
任务被取消！
当任务被取消时，会在协程内部抛出 CancelledError。
并发运行：asyncio.gather()、wait()、as_completed()
asyncio.gather()：批量并发运行任务
- 概念
asyncio.gather()是最常用的并发执行函数。
它的作用是并发执行多个协程或 Task，并在全部完成后返回结果列表。 - 语法：
results = await asyncio.gather(*aws, return_exceptions=False)aws：一组协程对象或Task对象。return_exceptions：False（默认） → 如果某个任务抛异常，会立即传播。True→ 异常会作为结果返回，不会中断整个 gather。
- 示例：并发执行 3 个网络请求（模拟）
运行结果（顺序可能不同）：import asyncio import random async def fetch_data(name): print(f"开始获取数据：{name}") await asyncio.sleep(random.uniform(1, 3)) # 模拟网络延迟 print(f"完成：{name}") return f"结果：{name}" async def main(): results = await asyncio.gather( fetch_data("用户信息"), fetch_data("订单信息"), fetch_data("库存信息") ) print("全部完成：", results) asyncio.run(main())
要点：开始获取数据：用户信息 开始获取数据：订单信息 开始获取数据：库存信息 完成：订单信息 完成：库存信息 完成：用户信息 全部完成： ['结果：用户信息', '结果：订单信息', '结果：库存信息']- 三个任务同时开始。
gather()会等待所有任务都完成，然后一次性返回结果列表。- 返回顺序与传入顺序一致，不是完成顺序。
- 注意事项
asyncio.gather()内部会自动创建Task。- 如果要在中途取消任务，可以直接：
task = asyncio.create_task(fetch_data("A")) task.cancel() - 一般来说，
gather()是最推荐的批量并发方式，适合大部分异步并发需求。
asyncio.wait()：更灵活的等待控制
- 概念
asyncio.wait()与gather()类似，但：- 它不会直接返回结果；
- 它返回两个集合：
done：已完成的任务集合；pending：未完成的任务集合；
- 它可以设置超时时间或等待条件。
- 语法：
done, pending = await asyncio.wait(aws, timeout=None, return_when=asyncio.ALL_COMPLETED)参数 作用 aws可迭代对象（协程或任务） timeout超时时间（秒） return_when等待条件： ALL_COMPLETED、FIRST_COMPLETED、FIRST_EXCEPTION - 示例：等待最先完成的任务
输出结果：import asyncio import random async def job(name, sec): print(f"{name} 开始，预计 {sec} 秒") await asyncio.sleep(sec) print(f"{name} 完成") return name async def main(): tasks = [ asyncio.create_task(job("任务A", 3)), asyncio.create_task(job("任务B", 1)), asyncio.create_task(job("任务C", 2)) ] done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) print("已完成任务：", [t.result() for t in done]) for t in pending: t.cancel() asyncio.run(main())
要点：任务A 开始，预计 3 秒 任务B 开始，预计 1 秒 任务C 开始，预计 2 秒 任务B 完成 已完成任务： ['任务B']FIRST_COMPLETED表示只等最快完成的任务；- 其他任务仍在运行，你可以决定是否
cancel()； - 适合竞争请求、超时机制等场景。
asyncio.as_completed()：谁先完成谁先返回
- 概念
asyncio.as_completed()返回一个异步迭代器，
当某个协程完成时，会立刻产生一个结果（yield）。 - 语法：
for coro in asyncio.as_completed(aws): result = await coro - 示例：实时处理完成结果
运行输出（可能顺序不同）：import asyncio import random async def download_file(name): print(f"开始下载：{name}") await asyncio.sleep(random.uniform(1, 3)) print(f"下载完成：{name}") return name async def main(): tasks = [download_file(f"文件{i}") for i in range(1, 4)] for coro in asyncio.as_completed(tasks): result = await coro print(f"立即处理：{result}") asyncio.run(main())
要点：开始下载：文件1 开始下载：文件2 开始下载：文件3 下载完成：文件2 立即处理：文件2 下载完成：文件1 立即处理：文件1 下载完成：文件3 立即处理：文件3- 不会等全部任务完成；
- 谁先完成，谁的结果就先被处理；
- 适合流式处理、实时响应、日志聚合等。
异步 I/O 实战：网络请求与文件操作
为什么异步 I/O 能显著提升性能？
在传统同步代码中：
def download():
data = requests.get("https://example.com/data")
with open("data.txt", "w") as f:
f.write(data.text)
当 requests.get() 在等待服务器响应时，CPU 是闲置的。
若你有 100 个这样的下载任务，就得一个一个排队完成。
而在异步模型中（用 aiohttp + aiofiles）：
async def download(session, url):
async with session.get(url) as resp:
text = await resp.text()
async with aiofiles.open("data.txt", "w") as f:
await f.write(text)
当第一个请求在等待响应时，事件循环会切换到下一个请求。
CPU 不再浪费，所有请求能“并发”地被调度。
这就是异步 I/O 的核心优势。
异步文件操作（aiofiles）
-
概念简介
Python 的
open()是阻塞的。
若你用它写文件，会卡住整个事件循环。所以异步文件操作要使用专门的库：
aiofiles（标准异步文件库）。 -
安装
pip install aiofiles -
示例：
-
异步写文件
import asyncio import aiofiles async def write_file(): async with aiofiles.open("example.txt", "w") as f: await f.write("Hello, Async World!\n") await f.write("This is written asynchronously.") asyncio.run(write_file())解析：
语法 含义 async with aiofiles.open()异步上下文管理文件 await f.write()非阻塞写操作（不会卡住事件循环） -
异步读文件
import asyncio import aiofiles async def read_file(): async with aiofiles.open("example.txt", "r") as f: content = await f.read() print(content) asyncio.run(read_file())特点：
await f.read()是异步操作；- 可同时处理多个文件 I/O。
-
并发读写多个文件
import asyncio import aiofiles async def process_file(filename, text): async with aiofiles.open(filename, "w") as f: await f.write(text) async with aiofiles.open(filename, "r") as f: content = await f.read() print(f"{filename} content:", content) async def main(): tasks = [ asyncio.create_task(process_file("file1.txt", "Alpha")), asyncio.create_task(process_file("file2.txt", "Beta")), asyncio.create_task(process_file("file3.txt", "Gamma")), ] await asyncio.gather(*tasks) asyncio.run(main())所有文件操作几乎“同时”完成（在同一线程内异步调度）。
-
异步网络请求（aiohttp）
-
为什么不用
requests？requests是同步库，会阻塞事件循环。- 异步任务需要非阻塞的网络库。
aiohttp是异步 HTTP 客户端和服务端库。 -
安装
pip install aiohttp -
示例：
-
单个异步请求
import asyncio import aiohttp async def fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: text = await resp.text() print(f"Fetched {len(text)} characters from {url}") asyncio.run(fetch("https://example.com"))解析：
语法 说明 ClientSession()异步 HTTP 会话对象 session.get(url)异步 GET 请求 await resp.text()异步获取响应内容 -
并发抓取多个网页
import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as resp: text = await resp.text() print(f"{url} done, size={len(text)}") async def main(): urls = [ "https://example.com", "https://httpbin.org/get", "https://python.org", ] async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, url)) for url in urls] await asyncio.gather(*tasks) asyncio.run(main())同时发出所有请求，不等待前一个完成。
速度比同步requests快数倍。 -
下载网页并异步保存文件
import asyncio import aiohttp import aiofiles async def download_and_save(session, url): async with session.get(url) as resp: text = await resp.text() filename = url.split("//")[-1].replace("/", "_") + ".html" async with aiofiles.open(filename, "w") as f: await f.write(text) print(f"Saved {filename}") async def main(): urls = [ "https://example.com", "https://httpbin.org/get", "https://python.org", ] async with aiohttp.ClientSession() as session: tasks = [download_and_save(session, url) for url in urls] await asyncio.gather(*tasks) asyncio.run(main())
-
异步 I/O 性能分析
| 模式 | 请求方式 | 理论性能 |
|---|---|---|
同步 requests |
阻塞式 | 每次等待响应，CPU 空闲 |
多线程 requests |
并发，但受 GIL 限制 | 有上下文切换开销 |
异步 aiohttp |
单线程协作调度 | 高 I/O 并发，无锁、无切换负担 |
异步模型能极大地提升网络密集与文件密集任务的吞吐量。
异步上下文管理器、异步迭代器
异步上下文管理器（async with）基础
-
什么是上下文管理器？
在同步代码中：
with open("data.txt", "r") as f: data = f.read()它会自动执行两个步骤：
- 进入时调用
__enter__()； - 退出时调用
__exit__()（即使发生异常也会执行）。
而在异步世界中，我们也希望能：
- 异步地打开/关闭资源；
- 不阻塞事件循环。
于是 Python 3.5+ 引入了：
async with—— 异步上下文管理器语法。 - 进入时调用
-
异步上下文管理器的语法与机制
一个对象要能用于
async with，
必须实现两个特殊的异步方法：方法 作用 __aenter__(self)异步进入上下文时执行（对应 __enter__）__aexit__(self, exc_type, exc_val, exc_tb)异步退出上下文时执行（对应 __exit__） -
示例：
-
简单的异步上下文管理器
import asyncio class AsyncContext: async def __aenter__(self): print("进入上下文") await asyncio.sleep(1) return "异步资源就绪" async def __aexit__(self, exc_type, exc, tb): print("退出上下文") await asyncio.sleep(1) if exc: print("捕获到异常：", exc) async def main(): async with AsyncContext() as resource: print("使用资源：", resource) asyncio.run(main())输出：
进入上下文 使用资源： 异步资源就绪 退出上下文工作机制：
async with会调用__aenter__()获取资源；- 退出时自动调用
__aexit__()释放资源； - 整个过程不会阻塞事件循环。
-
在异步上下文中管理网络连接（实战）
import aiohttp import asyncio async def main(): async with aiohttp.ClientSession() as session: async with session.get("https://example.com") as resp: text = await resp.text() print(f"页面长度：{len(text)}") asyncio.run(main())aiohttp.ClientSession()就是一个典型的异步上下文管理器。
它内部会异步地：- 建立 HTTP 连接；
- 在退出时自动关闭 session；
- 保证没有资源泄漏。
-
自定义异步上下文管理器
-
为什么要自定义？
在复杂的异步程序中，可能需要：
- 打开/关闭数据库连接；
- 申请/释放异步锁；
- 管理异步资源（WebSocket、Session、File 等）。
这时就可以自定义异步上下文管理器。
-
示例：自定义异步数据库连接管理器
import asyncio class AsyncDB: async def __aenter__(self): print("正在连接数据库...") await asyncio.sleep(1) print("数据库连接成功") return self # 返回可操作对象 async def __aexit__(self, exc_type, exc_val, exc_tb): print("正在关闭数据库连接...") await asyncio.sleep(1) print("数据库连接已关闭") async def query(self, sql): print(f"执行查询：{sql}") await asyncio.sleep(0.5) return f"结果：[{sql}]" async def main(): async with AsyncDB() as db: result = await db.query("SELECT * FROM users") print(result) asyncio.run(main())输出：
正在连接数据库... 数据库连接成功 执行查询：SELECT * FROM users 结果：[SELECT * FROM users] 正在关闭数据库连接... 数据库连接已关闭
异步迭代器（async for）基础
-
普通 for 和异步 for 的区别
在同步代码中：
for x in [1, 2, 3]: print(x)在异步代码中（比如异步读取流数据）：
async for line in async_source: print(line)此时
async_source必须是一个 异步迭代器。 -
异步迭代器协议
一个对象能被
async for使用，必须定义以下方法：方法 说明 __aiter__(self)返回异步迭代器对象（通常返回 self）__anext__(self)异步返回下一个元素（使用 await） -
示例：自定义异步迭代器
import asyncio class AsyncCounter: def __init__(self, limit): self.limit = limit self.count = 0 def __aiter__(self): return self async def __anext__(self): if self.count < self.limit: await asyncio.sleep(0.5) self.count += 1 return self.count else: raise StopAsyncIteration async def main(): async for num in AsyncCounter(5): print(f"计数：{num}") asyncio.run(main())输出：
计数：1 计数：2 计数：3 计数：4 计数：5机制解析：
async for每次循环都会await __anext__()；- 当抛出
StopAsyncIteration时，循环结束； - 这样就能异步地产生序列。
异步生成器（async def + yield）
-
什么是异步生成器？
异步生成器是异步迭代器的语法糖，通过
async def+yield实现。 -
示例：异步生成器
import asyncio async def async_gen(): for i in range(3): await asyncio.sleep(1) yield i async def main(): async for value in async_gen(): print(f"收到值：{value}") asyncio.run(main())输出：
收到值：0 收到值：1 收到值：2特点：
- 可用
async for遍历； - 每次
yield前都可异步等待； - 可方便地用于异步流、WebSocket、日志流等场景。
- 可用
同步与异步的混合编程
为什么需要同步与异步混合？
在真实项目中，不可能所有代码都是异步的。
许多 Python 库（特别是第三方库）依然是 同步阻塞 的。
例如：
import requests # 同步网络库
import time
这些同步操作一旦运行，就会阻塞整个事件循环，导致异步性能完全失效。
示例：异步中混入同步阻塞代码
import asyncio
import requests
async def fetch_sync(url):
print(f"开始请求: {url}")
res = requests.get(url) # 同步阻塞！
print(f"完成: {url}, 状态码: {res.status_code}")
async def main():
await asyncio.gather(
fetch_sync("https://www.python.org"),
fetch_sync("https://www.baidu.com")
)
asyncio.run(main())
运行后结果：
开始请求: https://www.python.org
# 阻塞几秒……
完成: https://www.python.org, 状态码: 200
开始请求: https://www.baidu.com
完成: https://www.baidu.com, 状态码: 200
问题：两个任务不是并发执行，而是顺序执行。因为 requests.get() 是阻塞调用，会卡住事件循环。
所以：
- 异步程序中如果调用阻塞函数，整个 async 都失去意义。
- 但完全不用同步库也不现实，比如：
- 图像处理（Pillow）
- 爬虫（requests）
- 文件 I/O、数据库驱动、科学计算库等
所以需要一个中间方案：
“在异步中安全地运行同步函数” —— 这就是 线程池 + 异步混合编程。
在异步中运行同步任务 — loop.run_in_executor()
Python 的 asyncio 提供了一个非常强大的方法：
loop.run_in_executor(executor, func, *args)
作用：把一个同步函数（阻塞函数）放进线程池或进程池中异步运行。
这让我们能在异步代码中“包裹”同步操作，从而避免阻塞事件循环。
示例：异步中调用同步 requests
import asyncio
import requests
def fetch(url):
print(f"同步请求开始: {url}")
res = requests.get(url)
print(f"同步请求结束: {url}")
return res.status_code
async def main():
loop = asyncio.get_running_loop()
tasks = [
loop.run_in_executor(None, fetch, "https://www.python.org"),
loop.run_in_executor(None, fetch, "https://www.baidu.com"),
loop.run_in_executor(None, fetch, "https://www.bing.com"),
]
results = await asyncio.gather(*tasks)
print("所有任务完成:", results)
asyncio.run(main())
输出示例：
同步请求开始: https://www.python.org
同步请求开始: https://www.baidu.com
同步请求开始: https://www.bing.com
同步请求结束: https://www.python.org
同步请求结束: https://www.baidu.com
同步请求结束: https://www.bing.com
所有任务完成: [200, 200, 200]
说明：
- 我们没有改动
requests，它仍然是同步的； - 但借助
loop.run_in_executor，这些同步任务在 独立线程 中并行执行； asyncio等待它们的结果时不会被阻塞。
使用自定义线程池与进程池
为什么要自定义 Executor？
前面用的是：
loop.run_in_executor(None, func, *args)
其中的 None 表示使用 默认线程池（ThreadPoolExecutor）。
但在真实项目中，我们通常需要：
- 控制线程/进程数量；
- 区分 CPU 密集型任务与 I/O 密集型任务；
- 管理多个线程池或进程池；
- 控制生命周期（例如在程序退出前关闭）。
这时我们就需要 自己创建 Executor 对象。
ThreadPoolExecutor：线程池执行器
ThreadPoolExecutor 来自 concurrent.futures 模块。
from concurrent.futures import ThreadPoolExecutor
构造方法
ThreadPoolExecutor(max_workers=None, thread_name_prefix='')
| 参数 | 说明 |
|---|---|
max_workers |
最大线程数，默认与 CPU 数量有关 |
thread_name_prefix |
线程名前缀，用于调试 |
示例：在异步中使用自定义线程池
import asyncio
from concurrent.futures import ThreadPoolExecutor
import time
def blocking_task(n):
print(f"任务 {n} 开始")
time.sleep(2)
print(f"任务 {n} 结束")
return n * n
async def main():
loop = asyncio.get_running_loop()
# 自定义线程池
with ThreadPoolExecutor(max_workers=3, thread_name_prefix="worker") as pool:
tasks = [
loop.run_in_executor(pool, blocking_task, i)
for i in range(5)
]
results = await asyncio.gather(*tasks)
print("结果:", results)
asyncio.run(main())
输出：
任务 0 开始
任务 1 开始
任务 2 开始
任务 3 开始
任务 4 开始
任务 0 结束
任务 1 结束
任务 2 结束
任务 3 结束
任务 4 结束
结果: [0, 1, 4, 9, 16]
分析：
- 任务函数是同步阻塞的（
time.sleep(2)）； - 通过线程池并行执行；
asyncio事件循环不会被阻塞；- 我们可以灵活调整
max_workers控制并发量。
ProcessPoolExecutor：进程池执行器
用于 CPU 密集型任务（如计算、加密、压缩、图像处理）。
为什么不用线程？因为 Python 的 GIL 限制线程无法并行执行 CPU 任务。
而多进程能绕过 GIL，充分利用多核 CPU。
示例：进程池执行器
import asyncio
from concurrent.futures import ProcessPoolExecutor
import math
import os
def cpu_bound_task(x):
print(f"进程 {os.getpid()} 计算 {x}!")
return sum(math.sqrt(i) for i in range(10_000_00))
async def main():
loop = asyncio.get_running_loop()
with ProcessPoolExecutor(max_workers=4) as pool:
tasks = [
loop.run_in_executor(pool, cpu_bound_task, i)
for i in range(5)
]
results = await asyncio.gather(*tasks)
print("完成所有任务:", results)
asyncio.run(main())
输出：
进程 13104 计算 0!
进程 13105 计算 1!
进程 13106 计算 2!
进程 13107 计算 3!
进程 13104 计算 4!
完成所有任务: [31622.7..., 31622.7..., 31622.7..., 31622.7..., 31622.7...]
分析：
- 每个任务在独立的进程中运行；
- GIL 不再是瓶颈；
- 适合 CPU 密集型计算；
- 与异步事件循环完美协作。
在同步代码中运行异步任务
为什么要在同步代码中运行异步？
在真实项目中，可能遇到以下情况：
- 你的主程序是同步的；
- 但你想调用一个异步 API；
- 或者你的框架（如 Flask）本身不支持 async 函数。
例如：
async def fetch_data():
await asyncio.sleep(1)
return "数据获取完成！"
# 不能直接这样
result = fetch_data()
会得到一个 coroutine 对象，而不是结果。
所以我们需要一种方式，让同步代码能安全地执行异步任务。
方法一：asyncio.run() —— 运行整个异步程序
这是最常见、最推荐的方式。
它会创建一个事件循环，运行直到任务完成，然后关闭循环。
import asyncio
async def fetch_data():
print("开始获取数据……")
await asyncio.sleep(1)
print("数据获取完成！")
return {"status": "ok", "data": [1, 2, 3]}
def main():
result = asyncio.run(fetch_data())
print("同步层获得结果：", result)
main()
输出：
开始获取数据……
数据获取完成！
同步层获得结果： {'status': 'ok', 'data': [1, 2, 3]}
特点：
| 优点 | 缺点 |
|---|---|
| 使用简单 | 只能调用一次（全局事件循环只能运行一次） |
| 自动关闭事件循环 | 无法与已有的事件循环嵌套运行 |
注意：
asyncio.run() 不能在已有事件循环中再次调用。
比如在 Jupyter Notebook、FastAPI 里，你会得到这个错误：
RuntimeError: asyncio.run() cannot be called from a running event loop
方法二：loop.run_until_complete()
适用于你自己管理事件循环的情况（较底层）。
import asyncio
async def task():
await asyncio.sleep(1)
return "OK"
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(task())
loop.close()
print("结果:", result)
一般不用手动这样写，除非你在自定义框架或调度器中控制 loop。
方法三：asyncio.to_thread() —— 在异步中运行同步代码（反向的反向）
这个方法出现在 Python 3.9+，是一个更优雅的方式，把同步函数丢进线程池中执行。
它其实是 run_in_executor() 的简化形式。
示例：异步函数中调用同步任务
import asyncio
import time
def blocking_io():
print("开始执行阻塞操作")
time.sleep(2)
print("阻塞操作完成")
return 42
async def main():
print("异步任务开始")
result = await asyncio.to_thread(blocking_io)
print("异步任务结束，结果：", result)
asyncio.run(main())
输出：
异步任务开始
开始执行阻塞操作
阻塞操作完成
异步任务结束，结果： 42
分析：
to_thread()自动在默认线程池中运行同步函数；- 不需要显式调用
loop.run_in_executor()； - 写法更简洁，也不需要管理线程池对象。
综合示例：同步主程序调用异步爬虫任务
import asyncio
import requests
# 同步请求函数
def sync_fetch(url):
print(f"[同步函数] 开始请求 {url}")
res = requests.get(url)
print(f"[同步函数] 完成请求 {url}，状态码：{res.status_code}")
return res.status_code
# 异步包装函数
async def async_fetch(url):
# 使用 asyncio.to_thread 在异步中运行同步函数
return await asyncio.to_thread(sync_fetch, url)
# 同步主程序入口
def run_async_task():
urls = [
"https://www.baidu.com",
"https://www.python.org",
"https://www.bing.com"
]
results = asyncio.run(asyncio.gather(*(async_fetch(url) for url in urls)))
print("全部任务完成:", results)
if __name__ == "__main__":
run_async_task()
运行结果示例
[同步函数] 开始请求 https://www.baidu.com
[同步函数] 开始请求 https://www.python.org
[同步函数] 开始请求 https://www.bing.com
[同步函数] 完成请求 https://www.python.org，状态码：200
[同步函数] 完成请求 https://www.baidu.com，状态码：200
[同步函数] 完成请求 https://www.bing.com，状态码：200
全部任务完成: [200, 200, 200]
分析说明：
| 关键点 | 说明 |
|---|---|
requests.get() 是阻塞函数 |
所以不能直接 await |
asyncio.to_thread() |
将同步函数扔进线程池执行 |
asyncio.gather() |
并发运行多个异步任务 |
asyncio.run() |
在同步入口中启动事件循环 |
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
4
0- 0
已为社区贡献13条内容
所有评论(0)