企业官网建设流程全解析

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 9101

python 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.py

"""E07: tagged CPU summaries returned in original input order."""fromconcurrent.futuresimportProcessPoolExecutor,as_completedimportmultiprocessingfromcpu_jobsimportchecksumdefsolve(skus,rounds=1000):results=[None]*len(skus)withProcessPoolExecutor(max_workers=2,mp_context=multiprocessing.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()rows=solve(['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.py

"""Compare missed timer deadlines, not an imaginary speedup."""importasynciofromtimeimportperf_counter,sleepasyncdefprobe(blocking):"""观察心跳是否被阻塞，不把一次定时器输出当作性能保证。"""started=perf_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.py

"""The same top-level CPU function introduced in lesson 07."""importosdefchecksum(job):"""纯函数传入数据、返回摘要；顶层定义供spawn子进程重新导入。"""index,sku,rounds=job value=0for_inrange(rounds):forcharinsku:value=(value*33+ord(char))%1000000007return{'index':index,'sku':sku,'checksum':value,'pid':os.getpid()}

lessons/08/domain.py

"""Shared business rules copied into each independent lesson."""fromdataclassesimportasdict,dataclass SUPPLIERS=('atlas','boreal','cedar')classProviderFailure(Exception):def__init__(self,code):self.code=codesuper().__init__(code)classInvalidData(ProviderFailure):def__init__(self):super().__init__('invalid_data')defvalidate_quantity(quantity):"""在任何外部调用前拒绝非法数量；布尔值不能冒充整数。"""iftype(quantity)isnotintornot1<=quantity<=100:raiseValueError('quantity must be an integer from 1 to 100')@dataclass(frozen=True,slots=True)classQuote:supplier:strsku:strunit_cents:intshipping_cents:intstock:int@classmethoddefparse(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*quantity+self.shipping_cents}defsummarize(rows,failures,quantity):"""缺货是正常过滤；最终排名由业务键决定，与完成顺序无关。"""quotes=sorted((rowforrowinrowsifrow['stock']>=quantity),key=lambdarow:(row['total_cents'],row['supplier']))return{'sku':'KB-001','quantity':quantity,'quotes':quotes,'failures':sorted(failures,key=lambdarow:row['supplier']),'best':quotes[0]ifquoteselseNone}

lessons/08/gateway.py

"""One async attempt. The caller owns and shares the AsyncClient."""importhttpxfromdomainimportInvalidData,ProviderFailure,Quoteasyncdeffetch(client,base_url,supplier,scenario,quantity):"""共享调用方客户端；只转换预期远端故障，让取消和程序错误传播。"""try:response=awaitclient.get(f'{base_url}/quotes/{supplier}',params={'sku':'KB-001','scenario':scenario},timeout=httpx.Timeout(2.0))response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure('timeout')fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(f'http_{exc.response.status_code}')fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏，不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure('transport')fromexctry:data=response.json()exceptValueErrorasexc:raiseInvalidData()fromexcreturnQuote.parse(data,supplier,'KB-001').row(quantity)

lessons/08/main.py

importargparseimportasyncioimportjsonimportosfromtimeimportperf_counterimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarize,validate_quantityfromgatewayimportfetchasyncdefcollect(client,base_url,scenario='normal',quantity=2):"""同时等待三家；客户端生命周期由外层run负责。"""validate_quantity(quantity)asyncdefone(supplier):try:returnawaitfetch(client,base_url,supplier,scenario,quantity),NoneexceptProviderFailureasexc:returnNone,{'supplier':supplier,'code':exc.code}# gather按输入顺序返回；价格排名仍交给业务函数决定。outcomes=awaitasyncio.gather(*(one(name)fornameinSUPPLIERS))returnsummarize([rowforrow,errorinoutcomesifrowisnotNone],[errorforrow,errorinoutcomesiferrorisnotNone],quantity)asyncdefrun(scenario):base_url=os.getenv('QUOTEFLOW_PROVIDER_URL','http://127.0.0.1:9101')started=perf_counter()asyncwithhttpx.AsyncClient(trust_env=False)asclient:result=awaitcollect(client,base_url,scenario)result['elapsed_ms']=round((perf_counter()-started)*1000,2)print(json.dumps(result,ensure_ascii=False,indent=2))if__name__=='__main__':parser=argparse.ArgumentParser()parser.add_argument('--scenario',default='normal')asyncio.run(run(parser.parse_args().scenario))

lessons/08/test_lesson.py

importasyncioimportunittestimporthttpxfrommainimportcollectclassLessonTests(unittest.IsolatedAsyncioTestCase):asyncdeftest_broken_gzip_is_invalid_data_and_not_retried(self):calls=[]asyncdefhandler(request):name=request.url.path.rsplit('/',1)[-1]calls.append(name)ifname=='boreal':returnhttpx.Response(200,headers={'Content-Encoding':'gzip'},content=b'not a gzip stream')returnhttpx.Response(200,json=dict(supplier=name,sku='KB-001',unit_cents=100,shipping_cents=0,stock=10))asyncwithhttpx.AsyncClient(transport=httpx.MockTransport(handler))asclient:result=awaitcollect(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):active=peak=0asyncdefhandler(request):nonlocalactive,peak name=request.url.path.rsplit('/',1)[-1]active+=1peak=max(peak,active)try:awaitasyncio.sleep(0.02)returnhttpx.Response(200,json=dict(supplier=name,sku='KB-001',unit_cents=100,shipping_cents=0,stock=10))finally:active-=1asyncwithhttpx.AsyncClient(transport=httpx.MockTransport(handler))asclient:result=awaitcollect(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()