目录一、为什么需要协程1. 回顾并发模型2. 多线程的问题3. 引出协程二、什么是协程1. 协程基本概念2. 协程核心特点三、async1. async def 与协程函数2. 协程函数与协程对象3. 代码验证四、await1. await 的作用2. 可等待对象3. await 控制权交接流程4. 代码案例五、asyncio1. 事件循环工作原理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 控制权并开始运行除非该协程主动挂起并让出控制权否则没有任何外部力量可以强行打断它的执行。任务之间的切换完全建立在各协程自愿协作的基础上协程的优势极低的上下文切换开销 协程的切换仅涉及用户态栈帧和寄存器状态的保存与恢复不涉及内核调度器切换耗时通常仅为纳秒级内存空间利用率 一个操作系统线程默认需要占用数兆字节的内存而一个协程仅需占用数百字节到几千字节的内存空间空间消除数据竞态 由于协程在单线程内运行且控制权转移只发生在明确的主动让出点代码在不涉及异步让出的原子操作区域内无需额外加锁即可保持数据一致性三、async自引入 async 和 await 语法关键字以来Python 正式提供了对原生协程的支持。在编写异步程序时区分 协程函数 与 协程对象 是掌握异步编程的前提1. async def 与协程函数使用 async def 声明的函数被称为协程函数async def fetch_data(): print(开始获取数据...) return Data语法特征在普通函数声明前加上 async 修饰符该函数即被标记为协程函数协程函数内部可以使用 await 关键字挂起执行普通函数内部使用 await 会引发语法错误 SyntaxError2. 协程函数与协程对象初学者在使用异步编程时常犯的一个错误是直接调用协程函数并期望其立即执行。理解两者的概念差异至关重要协程函数通过 async def 定义的函数本身本质是一个包含异步执行逻辑的函数模板协程对象直接调用协程函数所返回的对象。直接调用协程函数不会触发函数体内部代码的执行而仅产生一个封装了代码状态的协程对象3. 代码验证以下代码展示了协程函数与协程对象的类型差异以及直接调用协程函数的行为import inspect # 1. 定义协程函数 async def my_coroutine(): print(协程内部代码被执行) return 42 if __name__ __main__: # 2. 检查协程函数的类型 print(fmy_coroutine 的类型: {type(my_coroutine)}) print(- * 40) # 3. 直接调用协程函数 coro_obj my_coroutine() # 4. 检查返回值的类型 print(fcoro_obj 的类型: {type(coro_obj)})输出结果运行说明在上述代码中控制台未打印 协程内部代码被执行。如果单独执行 my_coroutine() 而不将其交由事件循环驱动或进行 awaitPython 解释器还会抛出 RuntimeWarning 警告四、await在异步编程中仅通过 async def 定义协程函数是不够的。要让协程在遇阻时释放 CPU 控制权必须配合await关键字1. await 的作用await 是协程中用于挂起执行的核心指令当程序运行到 await 语句时挂起当前协程当前协程暂停向下执行保留当前的上下文与变量状态交出 CPU 控制权将控制权交还给事件循环使事件循环能够去调度其他就绪的协程运行等待结果并恢复被 await 的异步操作完成后事件循环会在适当时机唤醒该协程将结果返回给变量并从 await 的下一行代码继续执行语法约束await 必须在被 async def 标记的协程函数内部使用在普通同步函数中使用 await 会抛出 SyntaxError2. 可等待对象await语法后面必须是一个 可等待对象在 Python 异步生态中主要包含以下三种可等待对象协程对象直接调用 async def 函数返回的对象Task 对象通过 asyncio.create_task() 等方法包装的排期协程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 fResult-{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我们将在后续章节中详述五、asyncioasyncio 是专门用于编写单线程并发异步代码的核心模块。要灵活掌控异步编程需要深入理解其核心引擎——事件循环以及协程、任务与 Future 的关系1. 事件循环工作原理事件循环是 asyncio 的核心引擎。它是一个无限循环的程序负责追踪所有异步任务的状态、监听 I/O 事件并在合适的时间点调度协程的执行工作步骤注册协程转化为 Task 后注册到事件循环的任务队列中调度事件循环选一个就绪的 Task 开始执行切换当该 Task 遇到 await 主动让出 CPU 时事件循环将其挂起并立即切换去执行下一个就绪的 Task唤醒底层 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 fResult-{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_exceptionsFalse) - list - *aws接收任意数量的可等待对象。如果传入的是原始协程对象 gather() 会自动将其封装为 Task 并注册进事件循环 - return_exceptions False若其中任意一个任务抛出未捕获的异常gather() 会立即向外抛出该异常 其他尚未完成的任务依然会在后台继续运行不会被自动取消 True 若某个任务抛出异常不会拦截中断而是将捕获到的异常对象作为该任务的返回结果 存入最终的结果列表中 返回值 按照参数传入的初始顺序返回一个包含所有任务结果的列表无论各个任务实际完成的先后顺序如何3. 代码案例以下代码对比了顺序串行执行与使用 asyncio.gather() 进行批量并发的性能开销差异并演示 return_exceptionsTrue 的用法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(fURL-{url_id} 连接超时) print(f[{time.strftime(%X)}] 完成请求 URL-{url_id}) return fHTTP 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_exceptionsTrue 确保单个任务失败不影响全量结果收集 results await asyncio.gather(*tasks, return_exceptionsTrue) print(f场景 B 完成总耗时: {time.time() - start_time:.2f} 秒) # 检查并解析结果 print(\n--- gather 结果 ---) for idx, item in enumerate(results, start1): if isinstance(item, Exception): print(f任务-{idx} 运行失败: {item}) else: print(f任务-{idx} 运行成功: {item}) if __name__ __main__: asyncio.run(main())输出结果4. asyncio 方法对比方法适用场景结果返回异常处理顺序 await强上下文依赖任务后一个任务必须使用前一个任务的输出单个返回触发异常立即打断流程create_task()动态、不固定数量的背景独立任务需对 Task 单独 await需对每个 Task 捕获异常asyncio.gather()固定数量的批量同构 / 异构任务并发最常用列表可通过 return_exceptionsTrue 捕获局部异常七、综合案例为了直观体验协程在海量 I/O 任务中的性能优势我们通过一个图片批量下载案例分别使用同步顺序下载、多线程并发下载以及协程并发下载三种方式实现并对它们的执行耗时与资源开销进行对比为了保证代码无第三方库依赖且测试结果稳定可复现代码中使用 time.sleep 与 asyncio.sleep 模拟网络传输与下载 I/O 延迟每张图片下载延迟设为 0.5 秒共处理 10 张图片1. 代码实现import asyncio import time from concurrent.futures import ThreadPoolExecutor # 待下载的图片 URL 列表 IMAGE_URLS [fhttps://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 fDownloaded: {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_workers5) 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 fDownloaded: {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 张图片下载时的底层表现对比分析如下下载模式执行特征总耗时计算依据系统开销同步任务串行执行当前图片下载阻塞时CPU 完全处于闲置等待状态5.00 秒10 * 0.5s 5.0s内存占用低但 CPU 利用率极低多线程5 个工作线程并发运行受线程数限制10 个任务分 2 批处理1.00 秒(10 / 5 ) * 0.5s 1.0s存在线程创建开销与系统上下文切换开销协程单线程非阻塞并发10 个任务同时排期并挂起在事件循环中0.50 秒max(0.5s) 0.5s无系统级切换开销内存占用极小结论同步阻塞在面对批量 I/O 时性能极差耗时随任务数量线性增加多线程提升了并发度但受限于线程池最大线程数量难以无限制扩展协程在单线程内实现了全量并发耗时仅取决于单个最长 I/O 操作的响应时间展现出了极高的并发效率总结本章围绕 Python 协程与异步编程展开学习了协程的基本概念、核心特点以及协程函数和协程对象之间的区别。通过 async 和 await 关键字我们理解了协程在遇到等待任务时主动让出执行权、并在条件满足后继续恢复运行的工作方式。同时我们对比了多个任务顺序执行与异步并发执行的区别掌握了 asyncio.gather() 组织多个协程任务的方法并通过批量下载图片案例直观体会了协程在网络请求等 I/O 密集型场景中的效率优势至此《Python从入门到实战》基础系列正式完结。从最初的基础语法、流程控制和函数到数据容器、面向对象、模块与包再到迭代器、文件操作、多进程、多线程与协程我们已经较为系统地学习了 Python 核心语法与常用开发思想。这些知识不仅是继续学习 Python 的基础也是进入数据分析、Web 开发、自动化、人工智能等方向的重要起点接下来我们将正式进入Python 数据分析系列从 NumPy、Pandas 等核心工具开始学习数据读取、清洗、处理、统计分析与可视化逐步建立完整的数据分析知识体系