17、Python之异步IO

一、概述异步IOAsynchronous I/O是一种单线程并发模型适合IO密集型场景网络请求、数据库读写、文件读取。 核心思想遇到IO阻塞时不原地等待让出CPU先去处理其它就绪任务等IO完成后再回来继续执行。区分多线程操作系统调度线程切换有开销受GIL限制CPU密集收益低协程异步用户态调度切换开销极小单线程内实现并发适合大量IO等待任务适用场景爬虫、大量接口请求、异步数据库不适合CPU密集计算。二、协程协程coroutine可以理解为“可以暂停、恢复的函数”由程序自身控制切换不是操作系统线程。 Python 3.5 引入async / await语法是协程的标准写法。async def定义协程函数调用它不会直接执行代码而是返回一个协程对象await后面接可等待对象协程、Task、Future遇到await就挂起当前协程让出事件循环关键概念事件循环 event loop异步的调度器负责管理、调度所有协程任务Task对协程的封装把协程提交给事件循环并发执行可等待对象Awaitable协程、Task、Future一句话协程是任务本体事件循环是调度器Task用来并发运行多个协程。⚠️ 注意await只能写在async def函数内部普通同步函数不能用await。三、使用 asyncioasyncio是Python内置异步IO标准库提供事件循环、Task、同步原语等。1. 基础单协程示例import asyncio async def hello(): print(hello start) # 异步睡眠让出循环不会阻塞整个线程 await asyncio.sleep(1) print(hello end) async def main(): await hello() if __name__ __main__: asyncio.run(main())asyncio.run()自动创建事件循环、运行协程、关闭循环Python3.7推荐写法2. 并发执行多个协程asyncio.gatherimport asyncio async def say(name, delay): print(f{name} 开始等待 {delay}s) await asyncio.sleep(delay) print(f{name} 结束) async def main(): # gather 收集多个协程并发执行等待全部完成 task1 say(A, 2) task2 say(B, 1) await asyncio.gather(task1, task2) if __name__ __main__: asyncio.run(main())运行总耗时 ≈ 2s不是 3s证明是并发。3. 手动创建 Taskimport asyncio async def work(n): await asyncio.sleep(n) print(ftask {n} done) async def main(): # 创建Task立刻提交到事件循环调度 t1 asyncio.create_task(work(1)) t2 asyncio.create_task(work(2)) await t1 await t2 if __name__ __main__: asyncio.run(main())4. 异步与同步代码坑点同步阻塞函数time.sleep()、同步MySQL、requests会卡死事件循环解决方案改用异步库aiohttp、aiomysql或者asyncio.to_thread()放到子线程运行同步代码import asyncio import time def sync_block(): time.sleep(1) return 同步结果 async def main(): res await asyncio.to_thread(sync_block) print(res) asyncio.run(main())四、aiohttpaiohttp是基于asyncio的异步HTTP客户端/服务端框架替代同步的requests适合并发爬取网页、批量调用接口。安装pip install aiohttp1. 基础GET请求import asyncio import aiohttp async def fetch(url): # 创建会话类比 requests.Session async with aiohttp.ClientSession() as session: async with session.get(url) as resp: print(状态码, resp.status) # 读取文本 text await resp.text() return text[:100] async def main(): url [https://httpbin.org/get](https://httpbin.org/get) result await fetch(url) print(result) if __name__ __main__: asyncio.run(main())2. 批量并发请求import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as resp: return await resp.text() async def main(): urls [ [https://httpbin.org/get](https://httpbin.org/get), [https://httpbin.org/headers](https://httpbin.org/headers), [https://httpbin.org/ip](https://httpbin.org/ip) ] async with aiohttp.ClientSession() as session: tasks [fetch(session, u) for u in urls] results await asyncio.gather(*tasks) for r in results: print(r[:80]) if __name__ __main__: asyncio.run(main())最佳实践复用同一个 ClientSession不要每次请求新建Session减少连接开销。3. POST提交表单async def post_demo(session): data {username: test, pwd: 123456} async with session.post([https://httpbin.org/post](https://httpbin.org/post), datadata) as resp: return await resp.json()4. 常用配置超时timeoutaiohttp.ClientTimeout(total10)并发限制搭配asyncio.Semaphore控制并发数量防止请求打爆服务sem asyncio.Semaphore(3) # 最多同时3个请求 async def fetch_limit(session, url): async with sem: return await fetch(session, url)五、对比总结概念作用协程可暂停恢复的异步函数async def定义asyncio内置异步调度库提供事件循环、gather、Taskaiohttp异步HTTP库异步爬虫、接口调用首选六、常见坑混用requests asynciorequests是同步阻塞会卡住整个事件循环忘记 await协程对象不await代码不会执行也不会报错无限制并发请求触发对方服务器限流、封IP → 使用Semaphore限流不要在协程里面写耗时CPU计算异步不擅长这个