20 - 协程与异步编程

20 - 协程与异步编程这章讲 Python 的异步编程。说实话对入门来说偏深了但协程在现代 Python 生态里越来越重要FastAPI、httpx、aiohttp 都用至少得知道怎么回事。为什么需要异步假设你要请求 10 个网页。同步方式importrequests urls[fhttps://example.com/page/{i}foriinrange(10)]# 同步一个接一个请求forurlinurls:responserequests.get(url)# 每次都要等网络响应print(response.status_code)每个请求要等 0.5 秒的话10 个就要 5 秒。但实际上 CPU 在等网络的时候是空闲的——它在干等。异步方式可以在等待网络响应的时候去做别的事importasyncioimporthttpx# 异步 HTTP 客户端uv add httpxasyncdeffetch(url):asyncwithhttpx.AsyncClient()asclient:responseawaitclient.get(url)returnresponse.status_codeasyncdefmain():urls[fhttps://example.com/page/{i}foriinrange(10)]tasks[fetch(url)forurlinurls]resultsawaitasyncio.gather(*tasks)print(results)asyncio.run(main())10 个请求同时发出总耗时约 0.5 秒取决于最慢的那个。快了 10 倍。并发 vs 并行这两个概念很多人搞混先搞清楚并发Concurrency多个任务交替执行。一个 CPU 就够了任务之间轮流来。并行Parallelism多个任务同时执行。需要多个 CPU 核心。Python 的 asyncio 做的是并发不是并行。它在一个线程里让多个任务交替运行利用等待 I/O的空隙去做别的事。打个比方并发是一个厨师同时做几道菜这道菜炖着的时候去切那道菜的菜并行是好几个厨师各做一道菜。协程 vs 线程 vs 进程进程线程协程内存各自独立共享内存共享内存切换开销大系统级中极小用户级并发数量几十几百几万甚至几十万GIL 影响无受限不受限但只能在一个核上跑适用场景CPU 密集I/O 密集传统方式I/O 密集现代方式协程的优势轻量。一个线程里可以跑几万个协程内存和切换开销都极小。线程的切换由操作系统决定你不知道什么时候切协程的切换由你的代码控制在await的地方切。后者叫协作式多任务前者叫抢占式多任务。事件循环Event Loop协程的执行靠事件循环驱动。你可以把它理解成一个调度员事件循环 - A 协程在等网络响应好先挂起 A - 让 B 协程跑一会 - B 也在等数据库挂起 B - A 的网络响应回来了恢复 A - ...所有协程都在同一个线程里跑靠事件循环来切换。这就是为什么await那么重要——它是告诉事件循环我现在可以歇一歇你去忙别的的信号。协程基础importasyncio# 用 async def 定义协程函数asyncdefsay_hello():print(Hello)awaitasyncio.sleep(1)# 模拟 I/O 操作print(World)# 调用协程函数不会执行它而是返回一个协程对象corosay_hello()print(coro)# coroutine object say_hello at 0x...# 必须用事件循环来驱动它asyncio.run(coro)关键概念调用async def定义的函数不会立即执行它返回一个协程对象。你需要await它或者交给事件循环运行。这个跟生成器有点像——调用def gen(): yield 1也不会立即执行返回的是生成器对象。不是巧合后面会讲它们的关系。await 到底干了什么await做两件事等待后面的异步操作完成拿到结果交出控制权给事件循环让它去跑别的协程asyncdeffetch_data(url):print(f开始请求{url})responseawaithttpx.AsyncClient().get(url)# 等网络响应同时别人可以跑print(f收到响应{url})returnresponse.status_code如果await后面不是异步操作比如普通的time.sleep()那事件循环就被阻塞了所有协程都得等。这是新手最常犯的错importtimeasyncdefbad_example():time.sleep(5)# 错阻塞了整个事件循环# 应该用 await asyncio.sleep(5)原则异步函数里不要调用阻塞的同步函数。如果你不确定一个函数是不是阻塞的看它有没有async版本。并发执行多个协程方式一gather最常用importasyncioasyncdeftask(name,delay):print(f{name}开始)awaitasyncio.sleep(delay)print(f{name}完成)returnf{name}的结果asyncdefmain():resultsawaitasyncio.gather(task(A,2),task(B,1),task(C,3),)print(results)# [A 的结果, B 的结果, C 的结果]# 总耗时约 3 秒取最长的而不是 2136 秒asyncio.run(main())方式二create_task更灵活asyncdefmain():t1asyncio.create_task(task(A,2))t2asyncio.create_task(task(B,1))# 中间可以做别的事print(做点别的)# 等结果r1awaitt1 r2awaitt2print(r1,r2)create_task立即把协程调度到事件循环不用等await才开始。方式三as_completed谁先完成谁先来asyncdefmain():tasks[task(A,2),task(B,1),task(C,3)]forcoroinasyncio.as_completed(tasks):resultawaitcoroprint(f完成了一个{result})# 输出顺序B → A → C按完成时间适合完成一个处理一个的场景不用等所有任务都结束。超时和取消asyncdefmain():# 超时控制try:resultawaitasyncio.wait_for(task(A,10),timeout3)exceptasyncio.TimeoutError:print(超时了)# 取消任务tasyncio.create_task(task(B,10))awaitasyncio.sleep(1)t.cancel()try:awaittexceptasyncio.CancelledError:print(任务被取消了)超时和取消在实际项目中很重要——网络请求不能无限等用户关闭页面后后台任务应该取消。异步上下文管理器和异步迭代器async with跟普通的with一样但进入和退出可以是异步操作# 异步打开文件uv add aiofilesimportaiofilesasyncdefread_file_async(path):asyncwithaiofiles.open(path,r)asf:contentawaitf.read()returncontent# 异步 HTTP 客户端asyncdeffetch(url):asyncwithhttpx.AsyncClient()asclient:responseawaitclient.get(url)returnresponseasync for异步版的 for 循环每次迭代可以是异步的asyncdefstream_lines(path):asyncwithaiofiles.open(path,r)asf:asyncforlineinf:yieldline.strip()asyncdefmain():asyncforlineinstream_lines(big_file.txt):print(line)async for和async with只能在async def函数里使用。异步队列生产者-消费者模式这是异步编程里最经典的模式importasyncioasyncdefproducer(queue:asyncio.Queue,name:str):生产者往队列里放数据foriinrange(5):itemf{name}-{i}awaitqueue.put(item)print(f[生产者{name}] 放入{item})awaitasyncio.sleep(0.5)asyncdefconsumer(queue:asyncio.Queue,name:str):消费者从队列里取数据whileTrue:itemawaitqueue.get()print(f[消费者{name}] 取出{item})awaitasyncio.sleep(1)# 模拟处理时间queue.task_done()asyncdefmain():queueasyncio.Queue(maxsize10)# 2 个生产者 3 个消费者producers[asyncio.create_task(producer(queue,P1)),asyncio.create_task(producer(queue,P2)),]consumers[asyncio.create_task(consumer(queue,fC{i}))foriinrange(3)]awaitasyncio.gather(*producers)# 等生产者完成awaitqueue.join()# 等队列清空forcinconsumers:c.cancel()# 取消消费者asyncio.run(main())生产者产出数据消费者处理数据队列做缓冲。生产快消费慢的时候队列堆起来生产慢消费快的时候消费者等着。很灵活。异步中的异常处理asyncdefrisky_task():awaitasyncio.sleep(1)raiseValueError(出错了)asyncdefmain():# gather 里某个任务出错默认会直接抛异常try:awaitasyncio.gather(task(A,1),risky_task(),task(C,1),)exceptValueErrorase:print(f捕获到异常{e})# 用 return_exceptionsTrue 让出错的任务返回异常对象而不是抛出resultsawaitasyncio.gather(task(A,1),risky_task(),task(C,1),return_exceptionsTrue)forrinresults:ifisinstance(r,Exception):print(f任务失败{r})else:print(f任务成功{r})return_exceptionsTrue很实用——一个任务挂了不影响其他任务的结果收集。协程和生成器的关系第 17 章学了生成器yield其实协程就是从生成器演化来的# 生成器第 17 章defgen():yield1yield2# 老式协程Python 3.5 之前基于生成器asyncio.coroutinedefold_coro():yieldfromasyncio.sleep(1)# 现代协程Python 3.5用 async/awaitasyncdefnew_coro():awaitasyncio.sleep(1)async/await本质上是给生成器加了专门的语法让异步代码读起来更像同步代码。底层原理还是暂停 恢复——跟生成器的yield是一个思路。区别在于生成器yield暂停外部用next()恢复 → 数据生产者协程await暂停事件循环恢复 → 任务调度单元异步生成器async for用的和异步上下文管理器async with用的也是这个思路的延伸。同步代码和异步代码混用有时候你需要在异步代码里调用同步库比如requests或者反过来。异步里调同步run_in_executorimportasyncioimporttimeasyncdefmain():loopasyncio.get_event_loop()# 把阻塞操作扔到线程池里跑不阻塞事件循环resultawaitloop.run_in_executor(None,time.sleep,3)print(完成)asyncio.run(main())run_in_executor本质上是起了一个线程去跑阻塞代码。所以它其实是异步 多线程的混合。同步里调异步# 用 asyncio.run() 启动事件循环defsync_main():resultasyncio.run(some_async_function())print(result)注意asyncio.run()只能在没有正在运行的事件循环时调用。如果在 Jupyter Notebook 里已经有事件循环了要用await直接调用或者用nest_asyncio。什么时候用异步场景用什么网络请求爬虫、API 调用✅ 异步数据库查询✅ 异步用 asyncpg、motor 等异步驱动WebSocket✅ 异步文件读写⚠️ 可以用 aiofiles但提升不如网络明显CPU 密集计算❌ 用多进程multiprocessing简单脚本/工具❌ 同步就够了别给自己找麻烦异步编程的代码复杂度比同步高不少——调试更难、错误处理更复杂、不是所有库都有异步版本。小项目或脚本用同步就够了等到确实有性能需求的时候再上异步。一个实际的例子异步爬虫同时抓取多个页面importasyncioimporthttpximporttimeasyncdeffetch(session:httpx.AsyncClient,url:str)-dict:抓取单个页面try:responseawaitsession.get(url,timeout10)return{url:url,status:response.status_code,size:len(response.text),}exceptExceptionase:return{url:url,error:str(e)}asyncdefmain():urls[https://httpbin.org/delay/1,https://httpbin.org/delay/2,https://httpbin.org/delay/0,https://httpbin.org/status/404,https://httpbin.org/status/500,]starttime.time()asyncwithhttpx.AsyncClient()assession:tasks[fetch(session,url)forurlinurls]resultsawaitasyncio.gather(*tasks)elapsedtime.time()-startforrinresults:iferrorinr:print(f ❌{r[url]}—{r[error]})else:print(f ✅{r[url]}—{r[status]}({r[size]}bytes))print(f\n总耗时{elapsed:.2f}秒)# 同步方式大约需要 120... ≈ 3 秒# 异步方式约 2 秒取决于最慢的那个asyncio.run(main())本章小结并发是交替执行并行是同时执行。asyncio 做的是并发协程比线程更轻量一个线程里能跑几万个协程async def定义协程函数await暂停并交出控制权事件循环Event Loop是调度员驱动所有协程运行gather()并发执行多个协程create_task()更灵活as_completed()按完成顺序处理async with/async for是异步版的上下文管理器和迭代器不要在异步函数里调用阻塞的同步代码用run_in_executor扔到线程池协程从生成器演化而来底层都是暂停 恢复面试题Q1协程、线程、进程有什么区别Python 的 asyncio 属于哪种点击查看答案进程线程协程内存空间独立共享共享切换开销大系统级中极小用户级并发数量几十几百几万GIL不受影响受 GIL 限制在单线程内不涉及 GILasyncio 的协程是用户态的协作式多任务——所有协程跑在同一个线程里靠事件循环调度切换。切换由代码中的await触发协作式不像线程由操作系统抢占式调度。适合 I/O 密集型网络、数据库不适合 CPU 密集型应该用 multiprocessing。Q2await到底做了什么为什么不能在异步函数里调用time.sleep()点击查看答案await做两件事等待后面的 awaitable 对象完成获取结果挂起当前协程把控制权交还给事件循环让其他协程有机会运行time.sleep()是阻塞调用——它会卡住整个线程包括事件循环导致所有协程都被阻塞。# 错误阻塞事件循环asyncdefbad():time.sleep(5)# 所有协程都得等 5 秒# 正确让出控制权asyncdefgood():awaitasyncio.sleep(5)# 事件循环可以去跑别的协程同理requests.get()、open().read()大文件等同步 I/O 都会阻塞事件循环应该用异步替代httpx、aiofiles或run_in_executor。Q3asyncio.gather()和asyncio.create_task()有什么区别点击查看答案gather()一次性提交多个协程等全部完成后返回结果列表。更简洁适合一起跑、一起收的场景。create_task()立即将协程调度到事件循环返回 Task 对象。可以分别 await、取消、查状态。更灵活。# gather一起跑一起收resultsawaitasyncio.gather(coro1(),coro2(),coro3())# create_task先启动后收集t1asyncio.create_task(coro1())t2asyncio.create_task(coro2())# 中间可以做别的事r1awaitt1 r2awaitt2gather()内部其实就是对每个协程调用ensure_future()类似 create_task所以两者在并发行为上没有本质区别区别在于 API 的灵活度。Q4协程和生成器有什么关系点击查看答案协程是从生成器演化来的。两者核心机制相同暂停 恢复。生成器用yield暂停外部用next()恢复 →数据生产者协程用await暂停事件循环恢复 →任务调度单元Python 3.5 之前协程就是加了asyncio.coroutine装饰器的生成器用yield from实现异步。3.5 引入了async/await语法语义更清晰但底层原理一样。异步生成器async for和异步上下文管理器async with也是这个思路的延伸。