1. 为什么需要异步编程在传统的同步编程模型中代码执行是线性的 - 一个操作必须等待前一个操作完成后才能开始。这种阻塞式的执行方式在处理I/O密集型任务时效率极低因为大部分时间都在等待网络响应或磁盘读写。想象一下餐厅里只有一个服务员必须等前一个顾客点完餐才能服务下一位这就是同步模型的困境。异步编程通过非阻塞的方式解决了这个问题。它允许程序在等待某个操作完成时先去处理其他任务。就像餐厅有多个服务员一个在等厨房做菜时可以先去服务其他顾客。Python的asyncio库就是实现这种并发模型的利器。2. asyncio核心架构解析2.1 事件循环(Event Loop)事件循环是asyncio的核心引擎负责调度和执行协程任务。它不断检查两种队列准备就绪的任务队列I/O事件队列典型的事件循环工作流程从就绪队列获取任务执行直到遇到await将控制权交还事件循环检查I/O事件并唤醒等待中的任务重复上述过程import asyncio async def main(): print(Hello) await asyncio.sleep(1) print(World) # 获取事件循环并运行主协程 loop asyncio.get_event_loop() loop.run_until_complete(main())2.2 协程(Coroutine)协程是asyncio的基本执行单元通过async/await语法声明async def定义协程函数await暂停当前协程直到awaitable对象完成关键特性可暂停和恢复的执行保持自己的调用栈通过yield控制权而非线程切换2.3 Future与TaskFuture代表异步操作的最终结果而Task是Future的子类用于包装协程Future低级异步原语Task高级协程包装器都可用于await表达式async def fetch_data(): return data # 创建Task的三种方式 task1 asyncio.create_task(fetch_data()) task2 asyncio.ensure_future(fetch_data()) task3 asyncio.get_event_loop().create_task(fetch_data())3. 实战中的最佳实践3.1 正确管理事件循环新手常犯的错误是错误地创建多个事件循环。正确的做法应用层面通常只需要一个全局事件循环使用asyncio.run()Python 3.7避免混用不同循环# 正确方式 (Python 3.7) async def main(): pass asyncio.run(main()) # 传统方式 loop asyncio.get_event_loop() try: loop.run_until_complete(main()) finally: loop.close()3.2 任务并发控制不加限制地创建大量并发任务会导致资源耗尽。解决方案使用信号量(Semaphore)使用asyncio.gather()的return_exceptions参数实现任务队列async def worker(sem, url): async with sem: # 执行网络请求 pass async def main(): sem asyncio.Semaphore(10) # 最大并发数 tasks [worker(sem, url) for url in urls] await asyncio.gather(*tasks)3.3 错误处理模式异步代码的错误处理需要特别注意总是捕获并处理CancelledError为任务添加回调处理异常使用asyncio.shield保护关键任务async def critical_task(): try: # 关键操作 await asyncio.sleep(1) except asyncio.CancelledError: # 清理资源 raise async def main(): task asyncio.create_task(critical_task()) await asyncio.sleep(0.1) task.cancel() try: await task except asyncio.CancelledError: print(任务被取消)4. 性能优化技巧4.1 选择合适的I/O策略asyncio支持多种I/O实现默认选择器(SelectorEventLoop)ProactorEventLoop(Windows)uvloop(第三方高性能实现)# 使用uvloop提升性能 import uvloop uvloop.install()4.2 避免阻塞调用任何阻塞操作都会破坏事件循环使用loop.run_in_executor()包装阻塞调用特别注意文件I/O、CPU密集型计算避免在协程中使用time.sleep()async def read_large_file(): loop asyncio.get_event_loop() # 将阻塞调用转移到线程池 data await loop.run_in_executor(None, open(large.txt).read) return data4.3 监控与调试asyncio提供内置调试工具启用调试模式PYTHONASYNCIODEBUG1使用loop.slow_callback_duration第三方工具如aiomonitor# 设置慢回调警告阈值 loop asyncio.get_event_loop() loop.slow_callback_duration 0.1 # 100ms5. 常见问题与解决方案5.1 协程没有被执行常见原因忘记await协程调用没有通过asyncio.run()启动事件循环提前关闭检查清单确保所有协程调用都有await使用asyncio.run()作为入口点检查异常是否被静默处理5.2 任务卡死或无响应调试步骤检查是否有未处理的异常使用asyncio.wait_for设置超时检查是否有阻塞调用async def fetch_with_timeout(): try: await asyncio.wait_for(fetch_data(), timeout5.0) except asyncio.TimeoutError: print(请求超时)5.3 内存泄漏问题异步代码常见内存泄漏源未取消的任务引用回调函数保持对象引用全局变量积累解决方案定期清理完成的任务使用weakref处理回调监控内存使用async def monitor_memory(): while True: print(当前任务数:, len(asyncio.all_tasks())) await asyncio.sleep(60)6. 高级应用模式6.1 实现异步上下文管理器通过__aenter__和__aexit__支持async withclass AsyncConnection: async def __aenter__(self): self.conn await connect() return self.conn async def __aexit__(self, exc_type, exc, tb): await self.conn.close() async def main(): async with AsyncConnection() as conn: await conn.query(SELECT 1)6.2 使用异步生成器通过async for处理异步数据流async def async_range(count): for i in range(count): yield i await asyncio.sleep(0.1) async def main(): async for num in async_range(10): print(num)6.3 与多进程结合利用ProcessPoolExecutor实现CPU密集型并行async def cpu_bound_task(): loop asyncio.get_event_loop() with ProcessPoolExecutor() as pool: result await loop.run_in_executor( pool, cpu_intensive_function, args) return result7. 测试异步代码7.1 使用pytest-asyncioimport pytest pytest.mark.asyncio async def test_fetch_data(): data await fetch_data() assert data expected7.2 模拟异步依赖使用unittest.mock.AsyncMockfrom unittest.mock import AsyncMock async def test_service(): mock_db AsyncMock() mock_db.fetch.return_value mock_data service Service(mock_db) result await service.get_data() assert result mock_data7.3 基准测试使用asyncio内置工具测量性能async def benchmark(): start asyncio.get_event_loop().time() await asyncio.gather(*[task() for _ in range(1000)]) end asyncio.get_event_loop().time() print(f耗时: {end - start:.2f}秒) 
长河君 · 2026/9/12 22:40:27 · 阅读 1.2k Python异步编程与asyncio实战指南
1. 为什么需要异步编程在传统的同步编程模型中代码执行是线性的 - 一个操作必须等待前一个操作完成后才能开始。这种阻塞式的执行方式在处理I/O密集型任务时效率极低因为大部分时间都在等待网络响应或磁盘读写。想象一下餐厅里只有一个服务员必须等前一个顾客点完餐才能服务下一位这就是同步模型的困境。异步编程通过非阻塞的方式解决了这个问题。它允许程序在等待某个操作完成时先去处理其他任务。就像餐厅有多个服务员一个在等厨房做菜时可以先去服务其他顾客。Python的asyncio库就是实现这种并发模型的利器。2. asyncio核心架构解析2.1 事件循环(Event Loop)事件循环是asyncio的核心引擎负责调度和执行协程任务。它不断检查两种队列准备就绪的任务队列I/O事件队列典型的事件循环工作流程从就绪队列获取任务执行直到遇到await将控制权交还事件循环检查I/O事件并唤醒等待中的任务重复上述过程import asyncio async def main(): print(Hello) await asyncio.sleep(1) print(World) # 获取事件循环并运行主协程 loop asyncio.get_event_loop() loop.run_until_complete(main())2.2 协程(Coroutine)协程是asyncio的基本执行单元通过async/await语法声明async def定义协程函数await暂停当前协程直到awaitable对象完成关键特性可暂停和恢复的执行保持自己的调用栈通过yield控制权而非线程切换2.3 Future与TaskFuture代表异步操作的最终结果而Task是Future的子类用于包装协程Future低级异步原语Task高级协程包装器都可用于await表达式async def fetch_data(): return data # 创建Task的三种方式 task1 asyncio.create_task(fetch_data()) task2 asyncio.ensure_future(fetch_data()) task3 asyncio.get_event_loop().create_task(fetch_data())3. 实战中的最佳实践3.1 正确管理事件循环新手常犯的错误是错误地创建多个事件循环。正确的做法应用层面通常只需要一个全局事件循环使用asyncio.run()Python 3.7避免混用不同循环# 正确方式 (Python 3.7) async def main(): pass asyncio.run(main()) # 传统方式 loop asyncio.get_event_loop() try: loop.run_until_complete(main()) finally: loop.close()3.2 任务并发控制不加限制地创建大量并发任务会导致资源耗尽。解决方案使用信号量(Semaphore)使用asyncio.gather()的return_exceptions参数实现任务队列async def worker(sem, url): async with sem: # 执行网络请求 pass async def main(): sem asyncio.Semaphore(10) # 最大并发数 tasks [worker(sem, url) for url in urls] await asyncio.gather(*tasks)3.3 错误处理模式异步代码的错误处理需要特别注意总是捕获并处理CancelledError为任务添加回调处理异常使用asyncio.shield保护关键任务async def critical_task(): try: # 关键操作 await asyncio.sleep(1) except asyncio.CancelledError: # 清理资源 raise async def main(): task asyncio.create_task(critical_task()) await asyncio.sleep(0.1) task.cancel() try: await task except asyncio.CancelledError: print(任务被取消)4. 性能优化技巧4.1 选择合适的I/O策略asyncio支持多种I/O实现默认选择器(SelectorEventLoop)ProactorEventLoop(Windows)uvloop(第三方高性能实现)# 使用uvloop提升性能 import uvloop uvloop.install()4.2 避免阻塞调用任何阻塞操作都会破坏事件循环使用loop.run_in_executor()包装阻塞调用特别注意文件I/O、CPU密集型计算避免在协程中使用time.sleep()async def read_large_file(): loop asyncio.get_event_loop() # 将阻塞调用转移到线程池 data await loop.run_in_executor(None, open(large.txt).read) return data4.3 监控与调试asyncio提供内置调试工具启用调试模式PYTHONASYNCIODEBUG1使用loop.slow_callback_duration第三方工具如aiomonitor# 设置慢回调警告阈值 loop asyncio.get_event_loop() loop.slow_callback_duration 0.1 # 100ms5. 常见问题与解决方案5.1 协程没有被执行常见原因忘记await协程调用没有通过asyncio.run()启动事件循环提前关闭检查清单确保所有协程调用都有await使用asyncio.run()作为入口点检查异常是否被静默处理5.2 任务卡死或无响应调试步骤检查是否有未处理的异常使用asyncio.wait_for设置超时检查是否有阻塞调用async def fetch_with_timeout(): try: await asyncio.wait_for(fetch_data(), timeout5.0) except asyncio.TimeoutError: print(请求超时)5.3 内存泄漏问题异步代码常见内存泄漏源未取消的任务引用回调函数保持对象引用全局变量积累解决方案定期清理完成的任务使用weakref处理回调监控内存使用async def monitor_memory(): while True: print(当前任务数:, len(asyncio.all_tasks())) await asyncio.sleep(60)6. 高级应用模式6.1 实现异步上下文管理器通过__aenter__和__aexit__支持async withclass AsyncConnection: async def __aenter__(self): self.conn await connect() return self.conn async def __aexit__(self, exc_type, exc, tb): await self.conn.close() async def main(): async with AsyncConnection() as conn: await conn.query(SELECT 1)6.2 使用异步生成器通过async for处理异步数据流async def async_range(count): for i in range(count): yield i await asyncio.sleep(0.1) async def main(): async for num in async_range(10): print(num)6.3 与多进程结合利用ProcessPoolExecutor实现CPU密集型并行async def cpu_bound_task(): loop asyncio.get_event_loop() with ProcessPoolExecutor() as pool: result await loop.run_in_executor( pool, cpu_intensive_function, args) return result7. 测试异步代码7.1 使用pytest-asyncioimport pytest pytest.mark.asyncio async def test_fetch_data(): data await fetch_data() assert data expected7.2 模拟异步依赖使用unittest.mock.AsyncMockfrom unittest.mock import AsyncMock async def test_service(): mock_db AsyncMock() mock_db.fetch.return_value mock_data service Service(mock_db) result await service.get_data() assert result mock_data7.3 基准测试使用asyncio内置工具测量性能async def benchmark(): start asyncio.get_event_loop().time() await asyncio.gather(*[task() for _ in range(1000)]) end asyncio.get_event_loop().time() print(f耗时: {end - start:.2f}秒)