Python从入门到实战（十八）：协程与异步编程

目录

一、为什么需要协程

1. 回顾并发模型

2. 多线程的问题

3. 引出协程

二、什么是协程

1. 协程基本概念

2. 协程核心特点

三、async

1. async def 与协程函数

2. 协程函数与协程对象

3. 代码验证

四、await

1. await 的作用

2. 可等待对象

3. await 控制权交接流程

4. 代码案例

五、asyncio

1. 事件循环工作原理

2. asyncio.run()

3. asyncio.create_task()

4. Coroutine、Task、Future 三者的关系

5. 代码案例

六、多任务

1. 顺序串行与异步并发

2. asyncio.gather()

3. 代码案例

4. asyncio 方法对比

七、综合案例

1. 代码实现

2. 输出结果与分析

总结

一、为什么需要协程

在前两章中，我们分别探讨了多进程与多线程编程。这两种技术均依赖操作系统的调度机制来实现任务交替或并行。然而，在面对现代高并发、高频 I/O 交互的业务场景时，传统的多进程与多线程模型展现出了各自的技术瓶颈

为了更高效地利用系统资源，我们需要引入一种更轻量级的并发实现方式——协程与异步编程

1. 回顾并发模型

在深入协程之前，我们需要先归纳此前学习过的两种并发模型：

• 多进程：

  • 优势：拥有独立的虚拟内存空间，能够绕过 GIL，在多核 CPU 上实现真正意义上的并行

  • 劣势：进程创建与销毁开销极高，操作系统进行进程上下文切换成本高昂，内存占用大

• 多线程：

  • 优势：共享同一进程的地址空间，创建与切换开销远小于进程

  • 劣势：受限于 GIL，无法利用多核 CPU 处理计算密集型任务；在共享内存下，多线程读写容易引发竞态条件，需要依赖锁等同步原语，增加了代码复杂度与死锁风险

2. 多线程的问题

对于网络爬虫、API 服务端、微服务通信以及数据库查询等典型的I/O 密集型应用，程序的瓶颈并不在于 CPU 计算速度，而在于等待网络响应或磁盘读写

在传统多线程中，处理海量并发 I/O 任务时面临以下问题：

线程开销

操作系统为每个线程分配独立的栈空间。如果系统需要同时处理 10000 个并发 HTTP 请求，创建 10000 个线程将消耗数 GB 的内存资源，容易引发内存溢出

上下文切换损耗

操作系统对线程采用抢占式调度。当线程数量极为庞大时，操作系统内核需要频繁挂起当前线程并恢复下一个线程的寄存器与栈状态。此时 CPU 将大量算力浪费在线程切换本身，而非真正的业务逻辑处理

大量 I/O 等待

网络传输和磁盘读写相较于 CPU 寄存器操作极其缓慢。在多线程等待 I/O 返回的过程中，线程会被操作系统挂起，CPU 处于长时间的阻塞闲置状态

3. 引出协程

为了解决 "并发 I/O 场景下线程开销过大且 CPU 资源被浪费" 的问题，协程被引入到 Python 中。协程，又称为 "微线程"。它不依赖操作系统内核进行调度，而是完全由程序在用户态自主控制

在协程模型中：

• 无操作系统切换开销：协程的切换仅在程序自身代码逻辑中进行，不涉及内核态与用户态的转换，极其轻量

• 高内存利用率：单个协程占用的内存空间极小，单个进程内可轻松创建数十万个协程

• 主动让出 CPU 控制权：当某个协程遇到 I/O 等待时，程序会主动暂停当前协程，将 CPU 控制权切换给其他就绪的协程；待 I/O 事件完成后再切回继续执行。这使得单线程即可最大程度发挥 CPU 的 I/O 处理能力

二、什么是协程

理解异步编程，首先需要掌握协程在计算机中的定义及其调度机制

1. 协程基本概念

协程是一种运行在用户态的轻量级并发执行单元

与普通的程序不同，普通函数只有一次调用和一次返回；而协程具备暂与恢复的能力。当协程在执行过程中遇到耗时的非阻塞操作时，可以主动暂停自身，将控制权交出，待 I/O 操作完成后再从暂停位置恢复执行

协程的核心在于利用单线程实现多任务的高效轮换，其基本概念包含以下要素：

• 用户态控制：操作系统内核感知不到协程的存在。协程的创建、调度和销毁完全由 Python 解释器和用户代码控制，不需要进行内核态与用户态的频繁切换

• 单线程高并发：一个线程内部可以存在成千上万个协程。所有协程共享同一个线程的地址空间与资源

• 非阻塞驱动：协程必须配合非阻塞 I/O 操作使用。当遇到 I/O 等待时，协程挂起自身，使 CPU 能够立即去处理其他就绪的协程

2. 协程核心特点

理解协程的技术本质，关键在于区分它与线程/进程在调度机制上的差异：

核心准则：协程不是抢占式调度，而是协作式调度

抢占式调度与协作式调度

• 抢占式调度： 应用于操作系统管理的进程与线程。操作系统通过硬件定时器中断与时间片机制，可以在任意时间点剥夺当前线程的 CPU 控制权，并切换给其他线程。开发者无法预测代码何时会被打断

• 协作式调度： 应用于协程。某个协程一旦获取 CPU 控制权并开始运行，除非该协程主动挂起并让出控制权，否则没有任何外部力量可以强行打断它的执行。任务之间的切换完全建立在各协程自愿协作的基础上

协程的优势

1. 极低的上下文切换开销： 协程的切换仅涉及用户态栈帧和寄存器状态的保存与恢复，不涉及内核调度器，切换耗时通常仅为纳秒级

2. 内存空间利用率： 一个操作系统线程默认需要占用数兆字节的内存，而一个协程仅需占用数百字节到几千字节的内存空间空间

3. 消除数据竞态： 由于协程在单线程内运行，且控制权转移只发生在明确的主动让出点，代码在不涉及异步让出的原子操作区域内无需额外加锁即可保持数据一致性

三、async

自引入 async 和 await 语法关键字以来，Python 正式提供了对原生协程的支持。在编写异步程序时，区分 "协程函数" 与 "协程对象" 是掌握异步编程的前提

1. async def 与协程函数

使用 async def 声明的函数被称为协程函数

async def fetch_data(): print("开始获取数据...") return "Data"

语法特征：

• 在普通函数声明前加上 async 修饰符，该函数即被标记为协程函数

• 协程函数内部可以使用 await 关键字挂起执行（普通函数内部使用 await 会引发语法错误 SyntaxError）

2. 协程函数与协程对象

初学者在使用异步编程时，常犯的一个错误是直接调用协程函数并期望其立即执行。理解两者的概念差异至关重要：

• 协程函数：通过 async def 定义的函数本身，本质是一个包含异步执行逻辑的函数模板

• 协程对象：直接调用协程函数所返回的对象。直接调用协程函数不会触发函数体内部代码的执行，而仅产生一个封装了代码状态的协程对象

3. 代码验证

以下代码展示了协程函数与协程对象的类型差异，以及直接调用协程函数的行为：

import inspect # 1. 定义协程函数 async def my_coroutine(): print("协程内部代码被执行") return 42 if __name__ == "__main__": # 2. 检查协程函数的类型 print(f"my_coroutine 的类型: {type(my_coroutine)}") print("-" * 40) # 3. 直接调用协程函数 coro_obj = my_coroutine() # 4. 检查返回值的类型 print(f"coro_obj 的类型: {type(coro_obj)}")

输出结果：

运行说明：

在上述代码中，控制台未打印 "协程内部代码被执行"。如果单独执行 my_coroutine() 而不将其交由事件循环驱动或进行 await，Python 解释器还会抛出 RuntimeWarning 警告

四、await

在异步编程中，仅通过 async def 定义协程函数是不够的。要让协程在遇阻时释放 CPU 控制权，必须配合await关键字

1. await 的作用

await 是协程中用于挂起执行的核心指令

当程序运行到 await 语句时：

1. 挂起当前协程：当前协程暂停向下执行，保留当前的上下文与变量状态

2. 交出 CPU 控制权：将控制权交还给事件循环，使事件循环能够去调度其他就绪的协程运行

3. 等待结果并恢复：被 await 的异步操作完成后，事件循环会在适当时机唤醒该协程，将结果返回给变量，并从 await 的下一行代码继续执行

语法约束：await 必须在被 async def 标记的协程函数内部使用，在普通同步函数中使用 await 会抛出 SyntaxError

2. 可等待对象

await语法后面必须是一个 "可等待对象"

在 Python 异步生态中，主要包含以下三种可等待对象：

1. 协程对象：直接调用 async def 函数返回的对象

2. Task 对象：通过 asyncio.create_task() 等方法包装的排期协程

3. Future 对象：底层的底层异步对象，代表一个未来可能完成的操作结果

# 错误示范：None/普通对象不可等待 await print("Hello") # 错误示范：同步阻塞函数不可等待 await time.sleep(1) # 正确示范：使用 asyncio 提供的异步非阻塞延迟可等待对象 await asyncio.sleep(1)

3. await 控制权交接流程

理解 await 的控制流是掌握异步调度的关键。下图展示了两个协程在 await 触发时的调度交接过程：

4. 代码案例

下面的代码展示了 await 挂起与恢复的执行顺序：

import asyncio import time async def fetch_api(api_id: int, delay: int) -> str: print(f"[{time.strftime('%X')}] 协程-{api_id}: 开始发送请求...") # await 挂起当前协程，将控制权还给事件循环，非阻塞等待 delay 秒 await asyncio.sleep(delay) print(f"[{time.strftime('%X')}] 协程-{api_id}: 成功获取响应数据") return f"Result-{api_id}" async def main(): print(f"[{time.strftime('%X')}] 主协程启动") # 依次 await 两个协程对象 res1 = await fetch_api(1, 2) res2 = await fetch_api(2, 1) print(f"获取结果: {res1}, {res2}") if __name__ == "__main__": asyncio.run(main())

输出结果：

在上述案例中，虽然我们使用了 await，但由于我们是按顺序对 res1 和 res2 进行单步 await，导致协程-2 必须等协程-1 完全执行完后才启动。要让多个协程并发运行，需要借助 Task 或 asyncio.gather，我们将在后续章节中详述

五、asyncio

asyncio 是专门用于编写单线程并发异步代码的核心模块。要灵活掌控异步编程，需要深入理解其核心引擎——事件循环，以及协程、任务与 Future 的关系

1. 事件循环工作原理

事件循环是 asyncio 的核心引擎。它是一个无限循环的程序，负责追踪所有异步任务的状态、监听 I/O 事件并在合适的时间点调度协程的执行

工作步骤：

1. 注册：协程转化为 Task 后注册到事件循环的任务队列中

2. 调度：事件循环选一个就绪的 Task 开始执行

3. 切换：当该 Task 遇到 await 主动让出 CPU 时，事件循环将其挂起，并立即切换去执行下一个就绪的 Task

4. 唤醒：底层 I/O 事件准备就绪，事件循环将挂起的 Task 重新标记为就绪状态，等待下一次轮询调度

2. asyncio.run()

从 Python 3.7 开始，标准库提供了最高层级的入口 APIasyncio.run()

asyncio.run(main_coroutine)

asyncio.run() 自动完成了事件循环全生命周期的底层管理：

• 自动创建一个全新的事件循环

• 将传入的主协程作为入口任务运行，直至其完成

• 自动关闭事件循环并回收线程与网络连接资源

3. asyncio.create_task()

如前所述，直接对协程对象逐个进行 await 只会引发串行同步等待。要实现真正意义上的多任务并发，必须使用asyncio.create_task()将协程包装为Task（任务）

task = asyncio.create_task(coro_obj)

• 核心作用：create_task() 会立即将传入的协程对象注册到事件循环中。即使尚未对其执行 await，事件循环也会在有空闲时直接开始运行该 Task

• 返回值：返回一个 asyncio.Task 对象。通过对 Task 对象调用 await 可以阻塞等待并获取其最终返回值

4. Coroutine、Task、Future 三者的关系

在 asyncio 架构中，这三个概念构成了异步编程的技术骨架：

• Coroutine（协程）：编写的具体异步业务代码，无法独立运行，必须挂载到事件循环

• Task（任务）：Future 的子类。它将协程封装并注册进事件循环，管理任务的状态切换，并提供取消任务等控制方法

• Future：表示异步操作的最终结果。它是一个底层的 "结果占位符"，初始状态为 Pending（未完成），当异步操作完成后状态变为 Finished（完成）并存入返回数据或异常

5. 代码案例

下面的代码展示了如何使用 create_task() 将多个任务并发排到事件循环中运行：

import asyncio import time async def async_worker(task_id: int, delay: int) -> str: print(f"[{time.strftime('%X')}] 任务-{task_id} 启动") await asyncio.sleep(delay) print(f"[{time.strftime('%X')}] 任务-{task_id} 完成") return f"Result-{task_id}" async def main(): print(f"[{time.strftime('%X')}] 主程序开始排期任务") task1 = asyncio.create_task(async_worker(1, 2)) task2 = asyncio.create_task(async_worker(2, 1)) # await 获取任务结果 res1 = await task1 res2 = await task2 print(f"汇总结果: {res1}, {res2}") if __name__ == "__main__": asyncio.run(main())

输出结果：

结果分析：

两个任务的总运行耗时约 2 秒（取决于耗时最长的任务 1），而不是串行执行的 3 秒（2s + 1s）。这证明任务 1 和任务 2 在事件循环的驱动下实现了非阻塞并发执行

六、多任务

在前面的章节中，我们学习了如何创建单个并发任务。但在实际开发中，我们通常需要同时下发几十甚至成千上万个异步任务，并对这些任务的结果进行集中汇总与异常处理

1. 顺序串行与异步并发

理解多任务管理，首先需要区分 "按顺序 await" 与 "批量异步并发" 的执行模式差异：

• 按顺序 await： 在代码中对协程依次执行 await，会导致后续任务必须等待前一个任务彻底完成后才能启动，丧失了并发性能

• 批量异步并发模式： 将所有协程同时注册到事件循环中，使其在发生 I/O 阻塞时自动让出 CPU 并在后台共同交替运行，总耗时仅取决于耗时最长的那一个任务

2. asyncio.gather()

asyncio.gather() 是 Python 中处理批量并发任务最核心的 API

方法签名

asyncio.gather(*aws, return_exceptions=False) -> list - *aws：接收任意数量的可等待对象。如果传入的是原始协程对象 gather() 会自动将其封装为 Task 并注册进事件循环 - return_exceptions： False：若其中任意一个任务抛出未捕获的异常，gather() 会立即向外抛出该异常 其他尚未完成的任务依然会在后台继续运行，不会被自动取消 True： 若某个任务抛出异常不会拦截中断，而是将捕获到的异常对象作为该任务的返回结果 存入最终的结果列表中 返回值： 按照参数传入的初始顺序返回一个包含所有任务结果的列表，无论各个任务实际完成的先后顺序如何

3. 代码案例

以下代码对比了顺序串行执行与使用 asyncio.gather() 进行批量并发的性能开销差异，并演示 return_exceptions=True 的用法：

import asyncio import time async def fetch_url(url_id: int, delay: int) -> str: """模拟异步网络请求，若 url_id 为 3 则故意触发异常""" print(f"[{time.strftime('%X')}] 启动请求 URL-{url_id}") await asyncio.sleep(delay) if url_id == 3: raise ValueError(f"URL-{url_id} 连接超时！") print(f"[{time.strftime('%X')}] 完成请求 URL-{url_id}") return f"HTTP 200: URL-{url_id}" async def main(): # 场景 A: 顺序串行执行 start_time = time.time() print("=== 场景 A: 顺序串行执行 ===") res_a1 = await fetch_url(1, 1) res_a2 = await fetch_url(2, 2) print(f"场景 A 完成，总耗时: {time.time() - start_time:.2f} 秒\n") # 场景 B: 使用 asyncio.gather start_time = time.time() print("=== 场景 B: 使用 asyncio.gather 并发执行 ===") # 构建 3 个不同耗时的异步任务（其中 URL-3 会抛出异常） tasks = [ fetch_url(1, 1), fetch_url(2, 2), fetch_url(3, 1) ] # 设置 return_exceptions=True 确保单个任务失败不影响全量结果收集 results = await asyncio.gather(*tasks, return_exceptions=True) print(f"场景 B 完成，总耗时: {time.time() - start_time:.2f} 秒") # 检查并解析结果 print("\n--- gather 结果 ---") for idx, item in enumerate(results, start=1): if isinstance(item, Exception): print(f"任务-{idx} 运行失败: {item}") else: print(f"任务-{idx} 运行成功: {item}") if __name__ == "__main__": asyncio.run(main())

输出结果：

4. asyncio 方法对比

七、综合案例

为了直观体验协程在海量 I/O 任务中的性能优势，我们通过一个图片批量下载案例，分别使用同步顺序下载、多线程并发下载以及协程并发下载三种方式实现，并对它们的执行耗时与资源开销进行对比

为了保证代码无第三方库依赖且测试结果稳定可复现，代码中使用 time.sleep 与 asyncio.sleep 模拟网络传输与下载 I/O 延迟（每张图片下载延迟设为 0.5 秒，共处理 10 张图片）

1. 代码实现

import asyncio import time from concurrent.futures import ThreadPoolExecutor # 待下载的图片 URL 列表 IMAGE_URLS = [f"https://example.com/images/photo_{i}.jpg" for i in range(1, 11)] # 模式 1: 同步顺序下载 def download_image_sync(url: str) -> str: # 模拟网络下载 I/O 延迟 time.sleep(0.5) return f"Downloaded: {url}" def run_sync(): start_time = time.time() results = [] for url in IMAGE_URLS: res = download_image_sync(url) results.append(res) elapsed = time.time() - start_time print(f"[同步顺序下载] 处理 10 张图片，总耗时: {elapsed:.2f} 秒") return elapsed # 模式 2: 多线程并发下载 def run_threadpool(): start_time = time.time() # 使用 5 个工作线程的线程池 with ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(download_image_sync, IMAGE_URLS)) elapsed = time.time() - start_time print(f"[多线程下载] 处理 10 张图片，总耗时: {elapsed:.2f} 秒") return elapsed # 模式 3: asyncio 协程并发下载 async def download_image_async(url: str) -> str: # 使用 asyncio.sleep 模拟非阻塞网络 I/O 延迟 await asyncio.sleep(0.5) return f"Downloaded: {url}" async def run_async(): start_time = time.time() # 构建协程任务列表 tasks = [download_image_async(url) for url in IMAGE_URLS] # 使用 asyncio.gather 批量并发调度 results = await asyncio.gather(*tasks) elapsed = time.time() - start_time print(f"[asyncio 协程] 处理 10 张图片，总耗时: {elapsed:.2f} 秒") return elapsed # 4. 主入口与性能对比测试 if __name__ == "__main__": print("开始进行 10 张图片批量下载性能测试...\n") t_sync = run_sync() t_thread = run_threadpool() t_async = asyncio.run(run_async())

2. 输出结果与分析

控制台输出结果：

三种方式在处理 10 张图片下载时的底层表现对比分析如下：

结论：

1. 同步阻塞在面对批量 I/O 时性能极差，耗时随任务数量线性增加

2. 多线程提升了并发度，但受限于线程池最大线程数量，难以无限制扩展

3. 协程在单线程内实现了全量并发，耗时仅取决于单个最长 I/O 操作的响应时间，展现出了极高的并发效率

总结

本章围绕 Python 协程与异步编程展开，学习了协程的基本概念、核心特点，以及协程函数和协程对象之间的区别。通过 async 和 await 关键字，我们理解了协程在遇到等待任务时主动让出执行权、并在条件满足后继续恢复运行的工作方式。同时，我们对比了多个任务顺序执行与异步并发执行的区别，掌握了 asyncio.gather() 组织多个协程任务的方法，并通过批量下载图片案例，直观体会了协程在网络请求等 I/O 密集型场景中的效率优势

至此，《Python从入门到实战》基础系列正式完结。从最初的基础语法、流程控制和函数，到数据容器、面向对象、模块与包，再到迭代器、文件操作、多进程、多线程与协程，我们已经较为系统地学习了 Python 核心语法与常用开发思想。这些知识不仅是继续学习 Python 的基础，也是进入数据分析、Web 开发、自动化、人工智能等方向的重要起点

接下来，我们将正式进入Python 数据分析系列，从 NumPy、Pandas 等核心工具开始，学习数据读取、清洗、处理、统计分析与可视化，逐步建立完整的数据分析知识体系