Python进阶与并发编程08一个线程如何等待三个HTTP请求前两课分别用线程重叠网络等待、用进程处理Python密集计算。现在回到报价查询如果系统同时等待很多短HTTP响应为每个等待都占用一个线程并不是唯一选择。asyncio把“等待某件事完成”交给事件循环管理让同一个线程在不同任务的可运行片段之间切换。本课只要求掌握一个可验证的事实协程在真正等待异步I/O时让出控制权事件循环才有机会推进其他协程。async def不会自动把普通阻塞函数变成异步函数await也不意味着任何位置都会立刻切换任务。理解这个区别就能解释许多“明明用了async整个服务还是卡住”的问题。E07答案让位置随任务旅行上一课E07要求spawn进程池逐项计算摘要并按原输入顺序返回。答案在answer_previous.py顶层CPU函数单独放在cpu_jobs.py。它预先创建results[None] * len(skus)建立future - index映射哪个任务先完成就写回哪个槽位。这种设计没有阻止并行也没有要求工作者按顺序完成只是让输出顺序回到调用方需要的顺序。[KB-003,KB-001,KB-002]最终仍保持这三个位置索引为0、1、2。若输入出现两个相同SKU也不会覆盖。父进程对每个Future调用result()异常不会被静默丢弃最后检查所有槽位都有结果便于发现调度代码漏项。调用协程函数并没有开始跑它执行one(atlas)时得到一个协程对象它描述待执行的异步过程但没有像普通函数那样跑到return。必须由外层await它或由create_task、gather、TaskGroup等把它交给事件循环。创建后遗忘的协程通常会在垃圾回收时产生“was never awaited”警告同时对应HTTP根本没有发生。一个常见误写是先await one(atlas)再await one(boreal)最后等待cedar。每次都等上一个结束才创建下一个仍然是串行流程。本课的asyncio.gather(*(one(name) for name in SUPPLIERS))把三个协程一起交给调度器然后等待这组三个结果。gather成功返回的列表按照传入顺序组织而不是完成顺序。我们依然在业务层排序因为输入供应商顺序不等于价格顺序。这里有三种顺序输入名单顺序、实际完成顺序、最终报价排名。即便其中两种在某次运行中碰巧相同也不能合并它们的语义。asyncio.run(run(...))只放在脚本最外层它创建并管理事件循环。已经在事件循环里的函数应当继续await不要嵌套调用asyncio.run。后面FastAPI会提供运行循环路由函数不能为每个请求另开一个asyncio.run。Client生命周期不能跟着每次小请求走同步版httpx.Client换成httpx.AsyncClient发请求的位置改成await client.get(...)外层资源管理改成async with。不只是多写一个await连接的打开、读写和关闭现在都要配合异步生命周期。本课在run里创建一个AsyncClient把同一个对象传给collect再传给三次fetch。不要把AsyncClient()写在每个供应商函数内部那样三家分别建立连接池重复配置和连接建立调用很多次后成本更明显。也不要在模块导入时创建一个全局客户端然后永不关闭或者跨多次互不关联的asyncio.run随意复用已经绑定到旧循环的异步资源。资源归属可以用一句话描述谁创建客户端谁负责在所有子任务完成后关闭它。collect使用传入客户端却不擅自关闭调用者以后可以在同一个生命周期内执行多次collect。第12课会把这个生命周期搬进Web应用启动与停止阶段。用心跳证明阻塞而不是背一句“不要阻塞”blocking_demo.py同时运行一个10毫秒后醒来的心跳和一个150毫秒的工作。阻塞版工作调用time.sleep(0.15)协作版调用await asyncio.sleep(0.15)。两版工作花的名义等待时间相同差别在于前者占着事件循环线程后者把等待交还给调度器。2026-10-09制作环境的一次实际输出为阻塞版心跳150.49ms后才被处理协作版为24.97ms。10毫秒只是请求的延迟不是操作系统保证的精确定时。我们看的是心跳是否被完整长等待挡住而不是要求每台机器都打印同一数字。如果在async函数里调用同步httpx.Client.get效果与阻塞sleep相似函数签名虽然是异步的调用期间仍占住循环线程。纯Python长计算也一样即使换成async def checksum而内部没有让出控制权也不会得到CPU并行。遗留的短阻塞I/O可以考虑await asyncio.to_thread(sync_function, ...)移到线程但那继承了第06课的限制取消等待它的协程不等于杀掉已经运行的线程。CPU计算是否使用进程池则要依据第07课的成本判断。asyncio不是另一种万能加速器它主要解决大量I/O等待的调度。预期故障可以成为结果编程错误仍需暴露内层one只捕获ProviderFailure返回(成功行, None)或(None, 失败行)。这样503、网络超时、坏数据不会直接打断整个gather其他供应商仍可以完成。最外层过滤两种结果并使用原有summarize库存不足依然只是正常过滤。若出现未预期的RuntimeError本课让它传播。这里有一个暂时存在、必须明确的限制默认gather在一个子任务异常时会把异常交给调用者但并不因为这个异常自动取消所有其他子任务。剩余任务可能还在运行而调用者已经开始退出客户端上下文。主程序最外层的asyncio.run在结束时还会清理任务但不能依赖这点为长期服务管理资源。我们不在本课通过捕获一切来遮住这个问题。下一课使用TaskGroup建立明确的任务所有权任务组离开之前组里的任务必须已经结束发生编程错误时其他任务被取消并等待收尾。学完gather后知道它的边界比把所有并发API混用更重要。异步网关还区分内容编码损坏坏gzip在await client.get读取响应时就可能触发httpx.DecodingError还没有走到JSON解析。它应被转成InvalidData放在宽泛RequestError捕获之前否则会误报transport让未来重试策略把确定的数据错误当作短暂网络失败。测试使用带gzip响应头的损坏字节检查分类与一次尝试同时保留其余供应商。异步网关给每个HTTP阶段2秒超时这仍不是整个查询总预算。选择2秒是为了让第09课的0.8秒总预算先接管slow场景而不是让HTTPX单次读取先超时。两种超时最后都有用途但从不同层次约束等待。运行时看业务结果也看资源行为先在另一终端保持供应商运行再从课程根目录执行。Linux使用python3.11替换python。python tools/provider_lab.py--port 9101python lessons/08/answer_previous.py python lessons/08/main.py python lessons/08/main.py--scenario broken python lessons/08/blocking_demo.py python-m unittest discover-s lessons/08-p test_lesson.py-v正常结果仍是boreal25900与atlas26400模型变化不应改变业务公式。本课单元测试把外部HTTP传输替换为会真正await的本地处理器记录活动数峰值3并在结束后断言活动数0。测试同时确认collect返回后调用者持有的客户端仍可使用。它证明协程重叠与所有权真实供应商命令再证明网络路径。不要拿本课elapsed_ms直接除以第05课的数字宣称性能倍数。本课外层包含客户端构造和关闭05课只计collect在某些Windows环境客户端初始化可能比本地HTTP等待更明显。第17课会复用一致的客户端和计时范围做公平比较。E08三家并行只创建一个客户端原题E08用一个async with httpx.AsyncClient(...) as client包住整组报价让atlas、boreal、cedar同时查询normal场景购买2件。每次供应商函数接收同一个client内部不能创建新client。结果最低价为boreal25900客户端关闭发生在全部查询完成后通过异步活动计数验证峰值为3结束后为0。下一课提供完整答案。完成后尝试解释每个await在等待什么HTTP响应、多个任务完成还是资源关闭。若无法指出一个await对应的等待对象很可能只是把关键字写进去了还没有建立执行模型。把协程想成能暂停的调用栈一个协程执行到await时会等待所依赖的操作准备好。需要等待的那段时间里当前调用栈的局部变量被保存线程可以去执行其他已准备好的任务。响应到达后事件循环让这个协程从暂停点继续所以请求代码仍能按“先请求、后解析”的顺序书写而不需要手写一长串回调函数。但await不是强制让出执行权的万能按钮。如果被等待对象已经完成代码可能继续执行而不发生实际等待。因此不能靠“循环里有await”证明循环公平也不能在两个共享状态读写之间随意插入await后仍假定中间状态无人能看到。判断是否存在竞争应找出真实的暂停点和其他任务能够观察到的状态。举一个贴近报价的例子先读取剩余容量等待一次网络查询再把容量减一。即使只有一个线程两个协程也可能先后读取同一个旧容量然后分别通过检查最终超出上限。问题是业务检查与修改跨过了await使用Semaphore把获取与归还绑定成资源协议才比手动检查数字可靠。第十一课会把这一点写成完整实验。异步边界应该向上传递到调用方如果底层fetch变成协程那么调用它的collect通常也需要变成协程直到应用最外层由事件循环驱动。试图在同步函数内部临时启动一个循环再把结果装回同步接口容易与已有循环冲突也会反复创建和关闭异步资源。接口形状的变化应当是明确设计而不是藏在某个工具函数里的魔法。这并不意味着所有业务函数都要写async。金额计算、库存过滤和排序没有需要等待的外部操作保持普通函数更清楚也更容易单独测试。当前domain模块在同步与异步版本之间几乎不变说明并发调度和业务规则已经分开。只有真正等待资源的边界需要异步接口。任务数量和连接数量仍然是不同限制创建三个协程只是允许三个逻辑流程并发推进底层是否同时建立三个连接还受客户端连接池配置和HTTP协议影响。连接池已满时协程可能在等待连接而不是等待供应商响应。活动Task很多也不意味着实际网络吞吐高可能只是排队多。当前课程只有固定三家所以先把任务全部交给gather足够直接。规模变大时要限制任务生产、请求出站和应用入站这几种限制属于不同层次。以后遇到“为什么已经异步了还很慢”应观察时间花在等待连接、等待远端、等待可运行线程还是CPU计算不能只数代码里有多少个await。协程等待通常不消耗CPU但仍会持有栈、局部变量和关联对象。因此等待便宜不等于等待免费长期挂起的大量任务仍应有数量上限和取消路径。本课完整源码异步网关完整列在下面不能直接导入05课的同步函数后只在外面加await。cpu_jobs.py仅被E07答案使用报价实验仍保持单个事件循环线程。lessons/08/answer_previous.pyE07: tagged CPU summaries returned in original input order.fromconcurrent.futuresimportProcessPoolExecutor,as_completedimportmultiprocessingfromcpu_jobsimportchecksumdefsolve(skus,rounds1000):results[None]*len(skus)withProcessPoolExecutor(max_workers2,mp_contextmultiprocessing.get_context(spawn))aspool:jobs{pool.submit(checksum,(i,sku,rounds)):ifori,skuinenumerate(skus)}forfutureinas_completed(jobs):results[jobs[future]]future.result()assertall(rowisnotNoneforrowinresults)serial[checksum((i,sku,rounds))fori,skuinenumerate(skus)]assert[row[checksum]forrowinresults][row[checksum]forrowinserial]returnresultsif__name____main__:multiprocessing.freeze_support()rowssolve([KB-003,KB-001,KB-002])assert[row[sku]forrowinrows][KB-003,KB-001,KB-002]assert[row[index]forrowinrows][0,1,2]print(rows)lessons/08/blocking_demo.pyCompare missed timer deadlines, not an imaginary speedup.importasynciofromtimeimportperf_counter,sleepasyncdefprobe(blocking):观察心跳是否被阻塞不把一次定时器输出当作性能保证。startedperf_counter()asyncdefheartbeat():awaitasyncio.sleep(0.01)returnround((perf_counter()-started)*1000,2)asyncdefwork():ifblocking:sleep(0.15)else:awaitasyncio.sleep(0.15)tick,_awaitasyncio.gather(heartbeat(),work())returntickasyncdefmain():print({blocking_tick_ms:awaitprobe(True),cooperative_tick_ms:awaitprobe(False)})if__name____main__:asyncio.run(main())lessons/08/cpu_jobs.pyThe same top-level CPU function introduced in lesson 07.importosdefchecksum(job):纯函数传入数据、返回摘要顶层定义供spawn子进程重新导入。index,sku,roundsjob value0for_inrange(rounds):forcharinsku:value(value*33ord(char))%1000000007return{index:index,sku:sku,checksum:value,pid:os.getpid()}lessons/08/domain.pyShared business rules copied into each independent lesson.fromdataclassesimportasdict,dataclass SUPPLIERS(atlas,boreal,cedar)classProviderFailure(Exception):def__init__(self,code):self.codecodesuper().__init__(code)classInvalidData(ProviderFailure):def__init__(self):super().__init__(invalid_data)defvalidate_quantity(quantity):在任何外部调用前拒绝非法数量布尔值不能冒充整数。iftype(quantity)isnotintornot1quantity100:raiseValueError(quantity must be an integer from 1 to 100)dataclass(frozenTrue,slotsTrue)classQuote:supplier:strsku:strunit_cents:intshipping_cents:intstock:intclassmethoddefparse(cls,data,supplier,sku):在HTTP边界把不可信JSON变成身份、金额和库存均可信的报价。ifnotisinstance(data,dict):raiseInvalidData()ifdata.get(supplier)!supplierordata.get(sku)!sku:raiseInvalidData()forkeyin(unit_cents,shipping_cents,stock):iftype(data.get(key))isnotintordata[key]0:raiseInvalidData()returncls(supplier,sku,data[unit_cents],data[shipping_cents],data[stock])defrow(self,quantity):return{**asdict(self),total_cents:self.unit_cents*quantityself.shipping_cents}defsummarize(rows,failures,quantity):缺货是正常过滤最终排名由业务键决定与完成顺序无关。quotessorted((rowforrowinrowsifrow[stock]quantity),keylambdarow:(row[total_cents],row[supplier]))return{sku:KB-001,quantity:quantity,quotes:quotes,failures:sorted(failures,keylambdarow:row[supplier]),best:quotes[0]ifquoteselseNone}lessons/08/gateway.pyOne async attempt. The caller owns and shares the AsyncClient.importhttpxfromdomainimportInvalidData,ProviderFailure,Quoteasyncdeffetch(client,base_url,supplier,scenario,quantity):共享调用方客户端只转换预期远端故障让取消和程序错误传播。try:responseawaitclient.get(f{base_url}/quotes/{supplier},params{sku:KB-001,scenario:scenario},timeouthttpx.Timeout(2.0))response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure(timeout)fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(fhttp_{exc.response.status_code})fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure(transport)fromexctry:dataresponse.json()exceptValueErrorasexc:raiseInvalidData()fromexcreturnQuote.parse(data,supplier,KB-001).row(quantity)lessons/08/main.pyimportargparseimportasyncioimportjsonimportosfromtimeimportperf_counterimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarize,validate_quantityfromgatewayimportfetchasyncdefcollect(client,base_url,scenarionormal,quantity2):同时等待三家客户端生命周期由外层run负责。validate_quantity(quantity)asyncdefone(supplier):try:returnawaitfetch(client,base_url,supplier,scenario,quantity),NoneexceptProviderFailureasexc:returnNone,{supplier:supplier,code:exc.code}# gather按输入顺序返回价格排名仍交给业务函数决定。outcomesawaitasyncio.gather(*(one(name)fornameinSUPPLIERS))returnsummarize([rowforrow,errorinoutcomesifrowisnotNone],[errorforrow,errorinoutcomesiferrorisnotNone],quantity)asyncdefrun(scenario):base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101)startedperf_counter()asyncwithhttpx.AsyncClient(trust_envFalse)asclient:resultawaitcollect(client,base_url,scenario)result[elapsed_ms]round((perf_counter()-started)*1000,2)print(json.dumps(result,ensure_asciiFalse,indent2))if__name____main__:parserargparse.ArgumentParser()parser.add_argument(--scenario,defaultnormal)asyncio.run(run(parser.parse_args().scenario))lessons/08/test_lesson.pyimportasyncioimportunittestimporthttpxfrommainimportcollectclassLessonTests(unittest.IsolatedAsyncioTestCase):asyncdeftest_broken_gzip_is_invalid_data_and_not_retried(self):calls[]asyncdefhandler(request):namerequest.url.path.rsplit(/,1)[-1]calls.append(name)ifnameboreal:returnhttpx.Response(200,headers{Content-Encoding:gzip},contentbnot a gzip stream)returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock10))asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(handler))asclient:resultawaitcollect(client,http://lab)self.assertEqual(result[failures],[{supplier:boreal,code:invalid_data}])self.assertEqual([row[supplier]forrowinresult[quotes]],[atlas,cedar])self.assertEqual(calls.count(boreal),1)asyncdeftest_one_client_can_overlap_three_real_coroutines(self):activepeak0asyncdefhandler(request):nonlocalactive,peak namerequest.url.path.rsplit(/,1)[-1]active1peakmax(peak,active)try:awaitasyncio.sleep(0.02)returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock10))finally:active-1asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(handler))asclient:resultawaitcollect(client,http://lab)self.assertFalse(client.is_closed)self.assertEqual(peak,3)self.assertEqual(active,0)self.assertEqual([r[supplier]forrinresult[quotes]],[atlas,boreal,cedar])if__name____main__:unittest.main() 
长河君 · 2026/10/17 2:38:54 · 阅读 1.2k Python进阶与并发编程08：一个线程如何等待三个HTTP请求
Python进阶与并发编程08一个线程如何等待三个HTTP请求前两课分别用线程重叠网络等待、用进程处理Python密集计算。现在回到报价查询如果系统同时等待很多短HTTP响应为每个等待都占用一个线程并不是唯一选择。asyncio把“等待某件事完成”交给事件循环管理让同一个线程在不同任务的可运行片段之间切换。本课只要求掌握一个可验证的事实协程在真正等待异步I/O时让出控制权事件循环才有机会推进其他协程。async def不会自动把普通阻塞函数变成异步函数await也不意味着任何位置都会立刻切换任务。理解这个区别就能解释许多“明明用了async整个服务还是卡住”的问题。E07答案让位置随任务旅行上一课E07要求spawn进程池逐项计算摘要并按原输入顺序返回。答案在answer_previous.py顶层CPU函数单独放在cpu_jobs.py。它预先创建results[None] * len(skus)建立future - index映射哪个任务先完成就写回哪个槽位。这种设计没有阻止并行也没有要求工作者按顺序完成只是让输出顺序回到调用方需要的顺序。[KB-003,KB-001,KB-002]最终仍保持这三个位置索引为0、1、2。若输入出现两个相同SKU也不会覆盖。父进程对每个Future调用result()异常不会被静默丢弃最后检查所有槽位都有结果便于发现调度代码漏项。调用协程函数并没有开始跑它执行one(atlas)时得到一个协程对象它描述待执行的异步过程但没有像普通函数那样跑到return。必须由外层await它或由create_task、gather、TaskGroup等把它交给事件循环。创建后遗忘的协程通常会在垃圾回收时产生“was never awaited”警告同时对应HTTP根本没有发生。一个常见误写是先await one(atlas)再await one(boreal)最后等待cedar。每次都等上一个结束才创建下一个仍然是串行流程。本课的asyncio.gather(*(one(name) for name in SUPPLIERS))把三个协程一起交给调度器然后等待这组三个结果。gather成功返回的列表按照传入顺序组织而不是完成顺序。我们依然在业务层排序因为输入供应商顺序不等于价格顺序。这里有三种顺序输入名单顺序、实际完成顺序、最终报价排名。即便其中两种在某次运行中碰巧相同也不能合并它们的语义。asyncio.run(run(...))只放在脚本最外层它创建并管理事件循环。已经在事件循环里的函数应当继续await不要嵌套调用asyncio.run。后面FastAPI会提供运行循环路由函数不能为每个请求另开一个asyncio.run。Client生命周期不能跟着每次小请求走同步版httpx.Client换成httpx.AsyncClient发请求的位置改成await client.get(...)外层资源管理改成async with。不只是多写一个await连接的打开、读写和关闭现在都要配合异步生命周期。本课在run里创建一个AsyncClient把同一个对象传给collect再传给三次fetch。不要把AsyncClient()写在每个供应商函数内部那样三家分别建立连接池重复配置和连接建立调用很多次后成本更明显。也不要在模块导入时创建一个全局客户端然后永不关闭或者跨多次互不关联的asyncio.run随意复用已经绑定到旧循环的异步资源。资源归属可以用一句话描述谁创建客户端谁负责在所有子任务完成后关闭它。collect使用传入客户端却不擅自关闭调用者以后可以在同一个生命周期内执行多次collect。第12课会把这个生命周期搬进Web应用启动与停止阶段。用心跳证明阻塞而不是背一句“不要阻塞”blocking_demo.py同时运行一个10毫秒后醒来的心跳和一个150毫秒的工作。阻塞版工作调用time.sleep(0.15)协作版调用await asyncio.sleep(0.15)。两版工作花的名义等待时间相同差别在于前者占着事件循环线程后者把等待交还给调度器。2026-10-09制作环境的一次实际输出为阻塞版心跳150.49ms后才被处理协作版为24.97ms。10毫秒只是请求的延迟不是操作系统保证的精确定时。我们看的是心跳是否被完整长等待挡住而不是要求每台机器都打印同一数字。如果在async函数里调用同步httpx.Client.get效果与阻塞sleep相似函数签名虽然是异步的调用期间仍占住循环线程。纯Python长计算也一样即使换成async def checksum而内部没有让出控制权也不会得到CPU并行。遗留的短阻塞I/O可以考虑await asyncio.to_thread(sync_function, ...)移到线程但那继承了第06课的限制取消等待它的协程不等于杀掉已经运行的线程。CPU计算是否使用进程池则要依据第07课的成本判断。asyncio不是另一种万能加速器它主要解决大量I/O等待的调度。预期故障可以成为结果编程错误仍需暴露内层one只捕获ProviderFailure返回(成功行, None)或(None, 失败行)。这样503、网络超时、坏数据不会直接打断整个gather其他供应商仍可以完成。最外层过滤两种结果并使用原有summarize库存不足依然只是正常过滤。若出现未预期的RuntimeError本课让它传播。这里有一个暂时存在、必须明确的限制默认gather在一个子任务异常时会把异常交给调用者但并不因为这个异常自动取消所有其他子任务。剩余任务可能还在运行而调用者已经开始退出客户端上下文。主程序最外层的asyncio.run在结束时还会清理任务但不能依赖这点为长期服务管理资源。我们不在本课通过捕获一切来遮住这个问题。下一课使用TaskGroup建立明确的任务所有权任务组离开之前组里的任务必须已经结束发生编程错误时其他任务被取消并等待收尾。学完gather后知道它的边界比把所有并发API混用更重要。异步网关还区分内容编码损坏坏gzip在await client.get读取响应时就可能触发httpx.DecodingError还没有走到JSON解析。它应被转成InvalidData放在宽泛RequestError捕获之前否则会误报transport让未来重试策略把确定的数据错误当作短暂网络失败。测试使用带gzip响应头的损坏字节检查分类与一次尝试同时保留其余供应商。异步网关给每个HTTP阶段2秒超时这仍不是整个查询总预算。选择2秒是为了让第09课的0.8秒总预算先接管slow场景而不是让HTTPX单次读取先超时。两种超时最后都有用途但从不同层次约束等待。运行时看业务结果也看资源行为先在另一终端保持供应商运行再从课程根目录执行。Linux使用python3.11替换python。python tools/provider_lab.py--port 9101python lessons/08/answer_previous.py python lessons/08/main.py python lessons/08/main.py--scenario broken python lessons/08/blocking_demo.py python-m unittest discover-s lessons/08-p test_lesson.py-v正常结果仍是boreal25900与atlas26400模型变化不应改变业务公式。本课单元测试把外部HTTP传输替换为会真正await的本地处理器记录活动数峰值3并在结束后断言活动数0。测试同时确认collect返回后调用者持有的客户端仍可使用。它证明协程重叠与所有权真实供应商命令再证明网络路径。不要拿本课elapsed_ms直接除以第05课的数字宣称性能倍数。本课外层包含客户端构造和关闭05课只计collect在某些Windows环境客户端初始化可能比本地HTTP等待更明显。第17课会复用一致的客户端和计时范围做公平比较。E08三家并行只创建一个客户端原题E08用一个async with httpx.AsyncClient(...) as client包住整组报价让atlas、boreal、cedar同时查询normal场景购买2件。每次供应商函数接收同一个client内部不能创建新client。结果最低价为boreal25900客户端关闭发生在全部查询完成后通过异步活动计数验证峰值为3结束后为0。下一课提供完整答案。完成后尝试解释每个await在等待什么HTTP响应、多个任务完成还是资源关闭。若无法指出一个await对应的等待对象很可能只是把关键字写进去了还没有建立执行模型。把协程想成能暂停的调用栈一个协程执行到await时会等待所依赖的操作准备好。需要等待的那段时间里当前调用栈的局部变量被保存线程可以去执行其他已准备好的任务。响应到达后事件循环让这个协程从暂停点继续所以请求代码仍能按“先请求、后解析”的顺序书写而不需要手写一长串回调函数。但await不是强制让出执行权的万能按钮。如果被等待对象已经完成代码可能继续执行而不发生实际等待。因此不能靠“循环里有await”证明循环公平也不能在两个共享状态读写之间随意插入await后仍假定中间状态无人能看到。判断是否存在竞争应找出真实的暂停点和其他任务能够观察到的状态。举一个贴近报价的例子先读取剩余容量等待一次网络查询再把容量减一。即使只有一个线程两个协程也可能先后读取同一个旧容量然后分别通过检查最终超出上限。问题是业务检查与修改跨过了await使用Semaphore把获取与归还绑定成资源协议才比手动检查数字可靠。第十一课会把这一点写成完整实验。异步边界应该向上传递到调用方如果底层fetch变成协程那么调用它的collect通常也需要变成协程直到应用最外层由事件循环驱动。试图在同步函数内部临时启动一个循环再把结果装回同步接口容易与已有循环冲突也会反复创建和关闭异步资源。接口形状的变化应当是明确设计而不是藏在某个工具函数里的魔法。这并不意味着所有业务函数都要写async。金额计算、库存过滤和排序没有需要等待的外部操作保持普通函数更清楚也更容易单独测试。当前domain模块在同步与异步版本之间几乎不变说明并发调度和业务规则已经分开。只有真正等待资源的边界需要异步接口。任务数量和连接数量仍然是不同限制创建三个协程只是允许三个逻辑流程并发推进底层是否同时建立三个连接还受客户端连接池配置和HTTP协议影响。连接池已满时协程可能在等待连接而不是等待供应商响应。活动Task很多也不意味着实际网络吞吐高可能只是排队多。当前课程只有固定三家所以先把任务全部交给gather足够直接。规模变大时要限制任务生产、请求出站和应用入站这几种限制属于不同层次。以后遇到“为什么已经异步了还很慢”应观察时间花在等待连接、等待远端、等待可运行线程还是CPU计算不能只数代码里有多少个await。协程等待通常不消耗CPU但仍会持有栈、局部变量和关联对象。因此等待便宜不等于等待免费长期挂起的大量任务仍应有数量上限和取消路径。本课完整源码异步网关完整列在下面不能直接导入05课的同步函数后只在外面加await。cpu_jobs.py仅被E07答案使用报价实验仍保持单个事件循环线程。lessons/08/answer_previous.pyE07: tagged CPU summaries returned in original input order.fromconcurrent.futuresimportProcessPoolExecutor,as_completedimportmultiprocessingfromcpu_jobsimportchecksumdefsolve(skus,rounds1000):results[None]*len(skus)withProcessPoolExecutor(max_workers2,mp_contextmultiprocessing.get_context(spawn))aspool:jobs{pool.submit(checksum,(i,sku,rounds)):ifori,skuinenumerate(skus)}forfutureinas_completed(jobs):results[jobs[future]]future.result()assertall(rowisnotNoneforrowinresults)serial[checksum((i,sku,rounds))fori,skuinenumerate(skus)]assert[row[checksum]forrowinresults][row[checksum]forrowinserial]returnresultsif__name____main__:multiprocessing.freeze_support()rowssolve([KB-003,KB-001,KB-002])assert[row[sku]forrowinrows][KB-003,KB-001,KB-002]assert[row[index]forrowinrows][0,1,2]print(rows)lessons/08/blocking_demo.pyCompare missed timer deadlines, not an imaginary speedup.importasynciofromtimeimportperf_counter,sleepasyncdefprobe(blocking):观察心跳是否被阻塞不把一次定时器输出当作性能保证。startedperf_counter()asyncdefheartbeat():awaitasyncio.sleep(0.01)returnround((perf_counter()-started)*1000,2)asyncdefwork():ifblocking:sleep(0.15)else:awaitasyncio.sleep(0.15)tick,_awaitasyncio.gather(heartbeat(),work())returntickasyncdefmain():print({blocking_tick_ms:awaitprobe(True),cooperative_tick_ms:awaitprobe(False)})if__name____main__:asyncio.run(main())lessons/08/cpu_jobs.pyThe same top-level CPU function introduced in lesson 07.importosdefchecksum(job):纯函数传入数据、返回摘要顶层定义供spawn子进程重新导入。index,sku,roundsjob value0for_inrange(rounds):forcharinsku:value(value*33ord(char))%1000000007return{index:index,sku:sku,checksum:value,pid:os.getpid()}lessons/08/domain.pyShared business rules copied into each independent lesson.fromdataclassesimportasdict,dataclass SUPPLIERS(atlas,boreal,cedar)classProviderFailure(Exception):def__init__(self,code):self.codecodesuper().__init__(code)classInvalidData(ProviderFailure):def__init__(self):super().__init__(invalid_data)defvalidate_quantity(quantity):在任何外部调用前拒绝非法数量布尔值不能冒充整数。iftype(quantity)isnotintornot1quantity100:raiseValueError(quantity must be an integer from 1 to 100)dataclass(frozenTrue,slotsTrue)classQuote:supplier:strsku:strunit_cents:intshipping_cents:intstock:intclassmethoddefparse(cls,data,supplier,sku):在HTTP边界把不可信JSON变成身份、金额和库存均可信的报价。ifnotisinstance(data,dict):raiseInvalidData()ifdata.get(supplier)!supplierordata.get(sku)!sku:raiseInvalidData()forkeyin(unit_cents,shipping_cents,stock):iftype(data.get(key))isnotintordata[key]0:raiseInvalidData()returncls(supplier,sku,data[unit_cents],data[shipping_cents],data[stock])defrow(self,quantity):return{**asdict(self),total_cents:self.unit_cents*quantityself.shipping_cents}defsummarize(rows,failures,quantity):缺货是正常过滤最终排名由业务键决定与完成顺序无关。quotessorted((rowforrowinrowsifrow[stock]quantity),keylambdarow:(row[total_cents],row[supplier]))return{sku:KB-001,quantity:quantity,quotes:quotes,failures:sorted(failures,keylambdarow:row[supplier]),best:quotes[0]ifquoteselseNone}lessons/08/gateway.pyOne async attempt. The caller owns and shares the AsyncClient.importhttpxfromdomainimportInvalidData,ProviderFailure,Quoteasyncdeffetch(client,base_url,supplier,scenario,quantity):共享调用方客户端只转换预期远端故障让取消和程序错误传播。try:responseawaitclient.get(f{base_url}/quotes/{supplier},params{sku:KB-001,scenario:scenario},timeouthttpx.Timeout(2.0))response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure(timeout)fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(fhttp_{exc.response.status_code})fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure(transport)fromexctry:dataresponse.json()exceptValueErrorasexc:raiseInvalidData()fromexcreturnQuote.parse(data,supplier,KB-001).row(quantity)lessons/08/main.pyimportargparseimportasyncioimportjsonimportosfromtimeimportperf_counterimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarize,validate_quantityfromgatewayimportfetchasyncdefcollect(client,base_url,scenarionormal,quantity2):同时等待三家客户端生命周期由外层run负责。validate_quantity(quantity)asyncdefone(supplier):try:returnawaitfetch(client,base_url,supplier,scenario,quantity),NoneexceptProviderFailureasexc:returnNone,{supplier:supplier,code:exc.code}# gather按输入顺序返回价格排名仍交给业务函数决定。outcomesawaitasyncio.gather(*(one(name)fornameinSUPPLIERS))returnsummarize([rowforrow,errorinoutcomesifrowisnotNone],[errorforrow,errorinoutcomesiferrorisnotNone],quantity)asyncdefrun(scenario):base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101)startedperf_counter()asyncwithhttpx.AsyncClient(trust_envFalse)asclient:resultawaitcollect(client,base_url,scenario)result[elapsed_ms]round((perf_counter()-started)*1000,2)print(json.dumps(result,ensure_asciiFalse,indent2))if__name____main__:parserargparse.ArgumentParser()parser.add_argument(--scenario,defaultnormal)asyncio.run(run(parser.parse_args().scenario))lessons/08/test_lesson.pyimportasyncioimportunittestimporthttpxfrommainimportcollectclassLessonTests(unittest.IsolatedAsyncioTestCase):asyncdeftest_broken_gzip_is_invalid_data_and_not_retried(self):calls[]asyncdefhandler(request):namerequest.url.path.rsplit(/,1)[-1]calls.append(name)ifnameboreal:returnhttpx.Response(200,headers{Content-Encoding:gzip},contentbnot a gzip stream)returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock10))asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(handler))asclient:resultawaitcollect(client,http://lab)self.assertEqual(result[failures],[{supplier:boreal,code:invalid_data}])self.assertEqual([row[supplier]forrowinresult[quotes]],[atlas,cedar])self.assertEqual(calls.count(boreal),1)asyncdeftest_one_client_can_overlap_three_real_coroutines(self):activepeak0asyncdefhandler(request):nonlocalactive,peak namerequest.url.path.rsplit(/,1)[-1]active1peakmax(peak,active)try:awaitasyncio.sleep(0.02)returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock10))finally:active-1asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(handler))asclient:resultawaitcollect(client,http://lab)self.assertFalse(client.is_closed)self.assertEqual(peak,3)self.assertEqual(active,0)self.assertEqual([r[supplier]forrinresult[quotes]],[atlas,boreal,cedar])if__name____main__:unittest.main()