Tianyao' s BLOG

Tianyao

0%

继续聊聊异步

发表于
本文字数： 4.4k 阅读时长 ≈ 8 分钟

和之前的文章呼应下，继续介绍python的异步机制。

在之前的文章中（指路[https://atffang.github.io/2025/02/21/asyncio浅析/#more]），我们简单学习了python的asynicio机制，但当我回看这篇文章时，仍然觉得许多东西——特别是概念，没有讲清楚。因此有了这篇文章，我们将从底层机制开始介绍python中的异步操作。

为什么要异步？

中国人应该都听过华罗庚烧开水的故事。其中办法甲是洗好水壶，灌上凉水，放在火上；在等待水开的时间里，洗茶壶、洗茶杯、拿茶叶；等水开了，泡茶喝。等待某个事件完成（比如水开）时，如果一直干等，时间就被浪费了；如果能利用等待时间去做其他有意义的事情，效率就会更高。在程序设计中，也是如此：

当程序遇到“必须等待”的操作时，比如等待网络响应、磁盘读写、用户输入，如果我们像传统方式那样干等，就会浪费大量宝贵时间。而如果我们能在等待的间隙去做别的事情，整体运行效率就能大幅提升。

这就是 异步编程（Asynchronous Programming） 的核心思想。异步编程是一种程序设计方式，它允许程序在遇到耗时操作时“暂时挂起当前任务”，去执行其他任务，待耗时操作完成后再回来继续执行。这样，程序不会因为等待某个操作而“卡死”或闲置，特别适合处理大量 I/O 操作的场景。

协程

协程（Coroutine）是一种程序结构，允许函数在执行过程中暂停（挂起）并在未来某个时间恢复执行。它是一种“可暂停的函数”，实现了非抢占式的多任务协作。通俗来说，协程可以在执行中途“让出控制权”，让其他协程运行，等到时机合适再“继续执行”，而不是像线程那样被操作系统抢占。

Python中的协程，特别是用 async def 定义的异步协程，实际上是基于生成器（generator） 机制演化而来的。生成器是一种特殊的迭代器，它可以在执行过程中暂停，并在需要时恢复执行。生成器函数使用 yield 关键字来生成值，每次调用 yield 时，函数会暂停执行，并返回一个值给调用者。下次调用生成器时，函数会从上次暂停的地方继续执行。这种暂停等待执行的状态成为挂起，传统函数调用有自己的调用栈，调用结束后栈帧销毁，而挂起状态的函数并不会销毁内部变量和参数。

Python中，每次函数调用都会创建一个 帧对象（PyFrameObject，C语言结构体），它包含了函数执行的所有上下文信息：

  • 当前指令指针（程序计数器，PC），指示下一条要执行的字节码位置
  • 局部变量和参数
  • 操作数栈状态
  • 代码对象引用

生成器和协程本质上也是函数调用，只不过在挂起时不会销毁帧对象，而是把它冻结保存起来。当协程执行到await或者yield时，Python解释器暂停执行当前帧，保存当前指令指针位置（即当前字节码执行到哪一步了）。局部变量、操作栈以及执行环境的状态都保存在该帧对象内，这个帧对象作为生成器/协程对象的内部状态，存活在内存中。当事件循环或调度器认为条件满足（比如等待的异步I/O完成），它会调用生成器/协程的 .send() 或 .__await__() 方法，让Python解释器加载之前保存的帧对象状态，从断点处继续执行字节码。

事件循环

事件循环（Event Loop） 是异步编程的核心调度机制，它不断循环监视“任务队列”或“事件源”，并在任务准备好执行时唤醒它们。我们在之前的文章中已经提到，协程对象执行到 await 会挂起并返回控制权，而要继续执行，就必须有一个机制来等待 await 的异步任务完成，再恢复暂停的协程。这个“等待+恢复”的过程，由事件循环自动调度完成。事件循环由下面四个关键部分组成：

  • Task 队列（任务队列）
    一个 Task 是对协程的封装，用于注册到事件循环中，并自动驱动协程的执行流程。协程对象会被封装为 asyncio.Task 对象，并加入事件循环管理的队列。

  • Future 对象
    表示一个尚未完成的异步操作，类似于 JavaScript 的 Promise。Future是Task的父类，而await 的结果通常是 Future。

  • IO 多路复用器
    使用操作系统的 select/epoll/kqueue/IOCP 来等待 I/O 事件。每个 await 的对象会注册一个 I/O 事件（如 socket 可读、定时器到期），这些事件会被提交到 selector，操作系统会在事件准备好时通知事件循环，事件循环再调用回调函数恢复协程。

  • 调度器
    调度器是事件循环的大脑，它控制协程的暂停与恢复。协程本质是生成器（generator）的拓展，调度行为是调用其 send() 或 throw() 方法。协程首次执行相当于 coro.send(None)，遇到 await 时将控制权交还给事件循环，并返回一个 Future 对象。事件循环等待 Future 对象完成，然后调用 task._step() 方法，执行 coro.send(result) ，恢复协程执行直到下一个 await 或结束。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
+---------------------------+
| asyncio.Task |
| - 包含一个协程对象 |
+---------------------------+
↓
初始执行 coro.send(None)
↓
协程执行 → 遇到 await
↓
返回 Future，挂起协程
↓
Future 注册到底层 IO selector
↓
事件完成 → Future 被标记为 done
↓
唤醒 Task → 再次执行 coro.send(result)
↓
直至结束

示例

爬虫

任务目标：我们有 5 个网页，要同时请求它们的内容并提取标题。我们希望在不阻塞主线程的情况下完成所有请求。如果使用串行执行的方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import requests
  
urls = [
"https://1.com",
"https://2.com",
"https://3.com",
"https://4.com",
"https://5.com"
] 

def get_title(url):
	response = requests.get(url)
	return f"{url} - status: {response.status_code}"

for url in urls:
	print(get_title(url))

所有请求会一个接一个等。
如果使用异步方式（并发请求，非阻塞）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio
import aiohttp

async def get_title(session, url):
    async with session.get(url) as response:
        text = await response.text()
        return f"{url} - status: {response.status}"

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [get_title(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r)

asyncio.run(main())

处理大量文件上传

你在做一个网站，有很多用户同时上传文件（比如照片），你需要接收上传请求/将文件保存到磁盘或对象存储/同时记录用户上传日志（如用户名、时间、文件名）到数据库或日志文件中。
如果使用同步方式，所有请求会一个接一个等，用户上传文件时，其他用户只能等待:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time

def save_file(user, file_name):
    time.sleep(2)  # 模拟写文件
    print(f"[保存] 用户 {user} 上传了 {file_name}")

def log_upload(user, file_name):
    time.sleep(1)  # 模拟写日志
    print(f"[日志] 记录 {user} 上传了 {file_name}")

def handle_upload(user, file_name):
    save_file(user, file_name)
    log_upload(user, file_name)

users = [("Alice", "a.png"), ("Bob", "b.jpg"), ("Carol", "c.docx")]
for user, file in users:
    handle_upload(user, file)

如果使用异步方式，所有请求可以同时处理，用户上传文件时，其他用户可以继续上传。

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

async def save_file(user, file_name):
    await asyncio.sleep(2)  # 模拟异步写文件
    print(f"[保存] 用户 {user} 上传了 {file_name}")

async def log_upload(user, file_name):
    await asyncio.sleep(1)  # 模拟异步写日志
    print(f"[日志] 记录 {user} 上传了 {file_name}")

async def handle_upload(user, file_name):
    await save_file(user, file_name)
    await log_upload(user, file_name)

async def main():
    users = [("Alice", "a.png"), ("Bob", "b.jpg"), ("Carol", "c.docx")]
    tasks = [handle_upload(user, file) for user, file in users]
    await asyncio.gather(*tasks)

asyncio.run(main())