[Python3高阶编程] - 异步编程深度学习指南一: 基础知识

下面学习的内容涉及源代码 https://blog.csdn.net/andylin02/article/details/159652101?spm1001.2014.3001.5502一、为什么需要异步——从阻塞 I/O 到并发模型1.1 同步 vs 异步 I/O同步 I/O发起请求 → 线程阻塞 → 等待完成 → 继续执行如requests.get()异步 I/O发起请求 → 立即返回 → 执行其他任务 → 数据就绪时回调处理如aiohttp.get()1.2 异步的优势高并发单线程可同时处理数千个 I/O 操作如 Web 请求、数据库查询低资源消耗无需创建大量线程避免上下文切换开销适合 I/O 密集型场景网络、文件、数据库等⚠️ 注意异步不能加速 CPU 密集型任务此时用多进程二、核心概念解析2.1 协程Coroutine定义可暂停和恢复的函数声明方式async def fetch(url): return await aiohttp.get(url) # 这里只是示例调用fetch(...)返回一个协程对象未执行2.2 事件循环Event Loop作用调度协程、监听 I/O 事件、执行回调获取当前 looploop asyncio.get_event_loop() # 或 (Python 3.7) loop asyncio.get_running_loop()2.3 Future / Task对象说明Future表示“未来某个时刻会完成的操作”底层抽象Task对协程的封装自动调度执行推荐使用task asyncio.create_task(fetch(https://example.com)) result await task三、async/await 语法详解3.1 基本结构import asyncio async def main(): print(Start) await asyncio.sleep(1) # 模拟异步 I/O print(End) # 启动Python 3.7 asyncio.run(main())3.2 关键规则await只能在async def中使用await后必须是awaitable 对象协程、Task、Future不要在异步函数中调用阻塞函数如time.sleep,requests.get❌ 错误示例 async def bad(): time.sleep(2) # 阻塞整个事件循环✅ 正确做法 async def good(): await asyncio.sleep(2) # 非阻塞四、并发执行模式4.1 顺序执行无并发async def main(): await fetch(url1) await fetch(url2) # 等第一个完成后才开始4.2 并发执行推荐方式1asyncio.gather()async def main(): results await asyncio.gather( fetch(url1), fetch(url2), fetch(url3) )特点所有任务同时启动等待全部完成异常处理默认一个失败则全部取消可用return_exceptionsTrue方式2asyncio.create_task() awaitasync def main(): t1 asyncio.create_task(fetch(url1)) t2 asyncio.create_task(fetch(url2)) r1 await t1 r2 await t2特点更灵活可分批等待方式3限制并发数防爆from asyncio import Semaphore sem Semaphore(10) # 最多10个并发 async def safe_fetch(url): async with sem: return await fetch(url) # 提交1000个任务 tasks [safe_fetch(url) for url in urls] results await asyncio.gather(*tasks)五、异步上下文管理器与迭代器5.1 异步上下文管理器async with用于异步资源管理如数据库连接、HTTP 会话async def fetch_data(): async with aiohttp.ClientSession() as session: async with session.get(https://api.example.com) as resp: return await resp.json()5.2 异步迭代器async for用于异步生成数据流async def async_range(n): for i in range(n): await asyncio.sleep(0.1) yield i async def main(): async for x in async_range(5): print(x)六、与同步代码集成6.1 在异步中运行阻塞函数使用loop.run_in_executor()将阻塞任务交给线程池def blocking_io(): with open(/tmp/file.txt, rb) as f: return f.read() async def main(): loop asyncio.get_running_loop() data await loop.run_in_executor(None, blocking_io) print(data)None表示使用默认线程池ThreadPoolExecutor6.2 从同步代码调用异步函数# 不推荐仅用于顶层 asyncio.run(async_func()) # 或在已有 loop 中如 Jupyter await async_func()⚠️不要混用asyncio.run()多次会创建多个 loop七、常用异步库推荐领域库说明HTTP 客户端aiohttp支持客户端和服务端HTTP 客户端httpx更现代支持 HTTP/2数据库asyncpg(PostgreSQL)高性能数据库aiomysql/aiosqliteMySQL / SQLiteWeb 框架FastAPI基于 Starlette原生异步Web 框架QuartFlask 的异步版爬虫scrapy(viascrapy-poet)或自研aiohttpasyncio八、调试与测试8.1 调试技巧启用调试模式asyncio.run(main(), debugTrue)查看未完成的 Tasktasks asyncio.all_tasks() for t in tasks: print(t)8.2 单元测试使用pytest-asyncioimport pytest pytest.mark.asyncio async def test_fetch(): result await fetch(https://example.com) assert result is not None九、常见陷阱与最佳实践❌ 陷阱1忘记awaitasync def main(): fetch(url) # 协程对象被创建但未执行✅ 正确await fetch(url)❌ 陷阱2在异步函数中使用阻塞 I/Oasync def handler(): requests.get(url) # 阻塞整个事件循环✅ 正确用aiohttp或run_in_executor❌ 陷阱3滥用asyncio.run()在已有事件循环中如 Jupyter、Web 框架不要调用asyncio.run()改为直接await✅ 最佳实践I/O 操作必须异步化网络、文件、DBCPU 密集型任务交给ProcessPoolExecutor限制并发数防止打爆目标服务器或本地资源使用类型注解async def f() - str:优雅关闭资源async with十、底层原理简析进阶10.1 事件循环如何工作基于Reactor 模式使用系统调用如 Linux 的epoll、macOS 的kqueue监听 I/O 事件当 socket 可读/可写时回调对应协程继续执行10.2 协程 vs 线程特性协程线程切换开销极低用户态高内核态内存占用KB 级MB 级并发能力数万数百受 GIL 和内存限制编程复杂度需理解异步模型需处理锁和竞态十一、学习路径建议入门掌握async/await、asyncio.run()、gather实战写一个异步爬虫或 API 客户端进阶理解事件循环、Task 调度、异常传播精通阅读asyncio源码、实现自定义异步协议十二、推荐阅读官方文档https://docs.python.org/3/library/asyncio.html书籍《Python 异步编程实战》中文、《Using Asyncio in Python》Caleb Hattingh视频David Beazley 的 “Async IO in Python” 演讲YouTube✅ 总结异步编程不是银弹而是解决 I/O 并发问题的利器。用对场景I/O 密集型→ 性能飞跃用错场景CPU 密集型→ 得不偿失忽略规则阻塞调用、忘记 await→ 程序退化为同步掌握async/await你就能用 Python 写出媲美 Node.js、Go 的高并发服务动手建议立即尝试将一个同步爬虫改造成异步版本对比 QPS 提升效果