Python3 asyncio 模块

asyncio是 Python3.4 + 内置的异步 I/O 框架，核心是通过协程（Coroutine） 实现异步编程，相比多线程 / 多进程，异步编程在 I/O 密集型任务（如网络请求、数据库操作、文件读写）中更轻量、开销更低，且能充分利用 CPU 资源。本文结合可运行示例，帮你从 0 到 1 掌握asyncio的核心逻辑。

一、核心认知：异步编程与 asyncio 的定位

1. 同步 vs 异步：核心差异

• 同步编程：代码按顺序执行，遇到 I/O 操作（如网络请求）时阻塞，等待操作完成后再执行下一行（例：单线程逐一下载 10 个文件，总耗时 = 每个文件下载时间之和）；
• 异步编程：代码执行到 I/O 操作时，不阻塞等待，而是 “挂起” 当前任务，去执行其他任务，等 I/O 操作完成后再回来继续执行（例：同时发起 10 个文件下载请求，总耗时≈最慢的那个文件下载时间）。

2. asyncio 的核心优势

• 基于单线程（无线程切换开销），通过事件循环调度协程，资源占用远低于多线程；
• 原生支持异步 I/O，无需第三方库即可实现高并发；
• 语法简洁（async/await关键字），相比回调函数更易维护。

3. 核心概念

概念	作用
协程（Coroutine）	异步执行的函数，用async def定义，通过await挂起 I/O 操作
事件循环（Event Loop）	asyncio 的核心，负责调度协程、管理 I/O 事件、处理任务切换
Task	封装协程的可调度对象，用于并发执行多个协程
Future	表示异步操作的 “未来结果”，Task 是 Future 的子类
awaitable	可被await的对象（协程、Task、Future 等）

二、基础用法：协程与事件循环

1. 定义与运行协程（最基础）

协程是异步编程的核心，需通过async def定义，且必须在事件循环中运行。

示例 1：基础协程定义与运行

import asyncio# 定义协程函数（async def标识）
async def hello_world():print("开始执行协程")# await挂起协程，模拟I/O操作（如网络请求）await asyncio.sleep(1)  # 注意：不能用time.sleep（同步阻塞）print("协程执行完成")if __name__ == "__main__":# 方法1：Python3.7+推荐用法（自动创建/关闭事件循环）asyncio.run(hello_world())# 方法2：手动管理事件循环（Python3.6及以下）# loop = asyncio.get_event_loop()# loop.run_until_complete(hello_world())# loop.close()

 

输出结果（等待 1 秒后输出）：

 

开始执行协程
协程执行完成

 

2. 并发执行多个协程（核心场景）

单个协程无异步优势，asyncio的核心价值是并发执行多个协程，需通过asyncio.create_task()创建 Task。

示例 2：并发执行多个协程 

 

import asyncio
import time# 定义耗时的协程函数（模拟I/O操作）
async def task(name, delay):print(f"任务{name}开始，延迟{delay}秒")start = time.time()await asyncio.sleep(delay)  # 异步休眠（非阻塞）end = time.time()print(f"任务{name}完成，耗时{end-start:.2f}秒")async def main():# 创建Task（并发执行）task1 = asyncio.create_task(task("A", 2))task2 = asyncio.create_task(task("B", 1))# 等待所有Task完成await task1await task2if __name__ == "__main__":start_total = time.time()asyncio.run(main())end_total = time.time()print(f"总耗时：{end_total-start_total:.2f}秒")

 

输出结果（总耗时≈2 秒，而非 3 秒）：

任务A开始，延迟2秒
任务B开始，延迟1秒
任务B完成，耗时1.00秒
任务A完成，耗时2.00秒
总耗时：2.00秒

 

3. 关键说明

• asyncio.create_task()：将协程封装为 Task，立即加入事件循环调度（无需等待await）；
• await task：等待单个 Task 完成，若需等待多个，可使用await asyncio.gather(task1, task2)（更简洁）；
• asyncio.sleep()：异步休眠（非阻塞），替代同步的time.sleep()（后者会阻塞整个事件循环）。

三、核心组件：进阶同步与控制

1. asyncio.gather ()：批量等待协程 / Task

批量管理多个协程 / Task 时，asyncio.gather()比逐个await更简洁，且能收集所有返回值。

示例 3：gather 收集返回值 

import asyncioasync def get_data(num):await asyncio.sleep(1)return f"数据{num}"async def main():# 批量创建协程并执行results = await asyncio.gather(get_data(1),get_data(2),get_data(3))print("所有结果：", results)  # 输出：所有结果： ['数据1', '数据2', '数据3']if __name__ == "__main__":asyncio.run(main())

 

2. Semaphore：控制并发数

异步编程中若需限制并发数（如避免同时发起 1000 个网络请求压垮服务器），可使用asyncio.Semaphore。

示例 4：限制最大并发数为 2 

import asyncio# 创建信号量，最大并发数2
semaphore = asyncio.Semaphore(2)async def limited_task(name, delay):# 获取信号量（超过并发数则等待）async with semaphore:print(f"任务{name}开始执行")await asyncio.sleep(delay)print(f"任务{name}执行完成")async def main():# 同时创建5个任务，但最多2个并发tasks = [asyncio.create_task(limited_task(i, 1)) for i in range(5)]await asyncio.gather(*tasks)if __name__ == "__main__":asyncio.run(main())

 

3. Event：协程间事件通知

实现协程间的 “等待 - 通知” 机制，类似多线程的threading.Event。

示例 5：Event 通知协程继续执行 

import asyncioasync def wait_event(event):print("协程等待事件触发...")await event.wait()  # 阻塞，直到event被set()print("事件触发，协程继续执行")async def main():event = asyncio.Event()# 创建并启动等待事件的协程task = asyncio.create_task(wait_event(event))# 模拟延迟后触发事件await asyncio.sleep(2)print("主线程触发事件")event.set()await taskif __name__ == "__main__":asyncio.run(main())

 

四、实战场景：异步网络请求（最常用）

asyncio最典型的应用是异步网络请求，需结合异步 HTTP 库（如aiohttp），相比同步的requests，并发效率提升 10 倍以上。

前置准备：安装 aiohttp 

pip install aiohttp

 

示例 6：异步批量请求接口 

import asyncio
import aiohttp# 待请求的URL列表
urls = ["https://httpbin.org/get?name=1","https://httpbin.org/get?name=2","https://httpbin.org/get?name=3"
]async def fetch_url(session, url):"""异步请求单个URL"""try:async with session.get(url, timeout=10) as resp:data = await resp.json()print(f"请求{url}成功，返回数据：{data['args']}")return dataexcept Exception as e:print(f"请求{url}失败：{e}")return Noneasync def main():# 创建异步HTTP会话（复用连接，提升效率）async with aiohttp.ClientSession() as session:# 批量创建请求任务tasks = [asyncio.create_task(fetch_url(session, url)) for url in urls]# 等待所有任务完成并收集结果results = await asyncio.gather(*tasks)print("所有请求完成，有效结果数：", len([r for r in results if r]))if __name__ == "__main__":asyncio.run(main())

 

五、常见坑点与避坑指南

1. 误用同步阻塞代码

• 问题：在协程中使用time.sleep()、requests.get()等同步阻塞代码，会阻塞整个事件循环，导致异步失效；
• 解决方案：替换为异步版本（asyncio.sleep()、aiohttp），若必须使用同步代码，需通过loop.run_in_executor()放到线程池执行：
   

  import asyncio
  import time# 同步阻塞函数
  def sync_task():time.sleep(1)return "同步任务完成"async def main():loop = asyncio.get_running_loop()# 将同步函数放到线程池执行（非阻塞）result = await loop.run_in_executor(None, sync_task)print(result)asyncio.run(main())
  

   
   

2. 忘记 await 可等待对象

• 问题：定义协程 / 创建 Task 后，未加await，导致协程未执行；
• 示例错误代码：
   

  async def task():await asyncio.sleep(1)print("任务执行")asyncio.run(task())  # 正确：await了协程
  # 错误：仅创建Task，未await，程序直接退出
  async def main():asyncio.create_task(task())
  asyncio.run(main())
  

   
   
• 解决方案：所有协程 / Task 必须通过await或asyncio.gather()等待执行。

3. 协程嵌套层级过深导致未执行

• 问题：在函数中嵌套定义协程但未调用，或忘记将子协程加入事件循环；
• 解决方案：确保所有协程最终都被asyncio.run()、await或Task调度。

4. 未处理异常

• 问题：协程中的异常若未捕获，会导致整个事件循环终止；
• 解决方案：在协程内添加try-except，或通过future.exception()捕获 Task 异常。

总结

关键点回顾

1. 核心定位：asyncio适用于 I/O 密集型任务（网络、文件、数据库），CPU 密集型任务仍需用multiprocessing；
2. 基础用法：协程用async def定义，await挂起 I/O 操作，通过asyncio.run()运行，并发用create_task()+gather()；
3. 避坑核心：避免在协程中使用同步阻塞代码，所有可等待对象必须加await，异常需手动捕获；
4. 实战优选：异步网络请求用aiohttp，限制并发用Semaphore，批量任务用gather()。