Python进阶与并发编程06：让三个HTTP等待重叠，线程池到底管理什么

第05课的程序能够在一家供应商失败时保留其他结果，但它仍然逐个等待。atlas返回之前，boreal连请求都没有发出去。对于这种大部分时间花在网络等待上的任务，我们可以让多个线程分别等待；主线程收集完成的结果。业务公式、协议校验和最低价规则不需要改变。

本课要解决的不是“把for循环换一个库”这么简单的问题。请求完成的顺序开始不稳定，多个工作者可能修改同一个计数器，调用方超时后线程可能仍在运行。只有同时理解这些行为，线程池才是资源管理工具，而不只是看起来更快的语法。

E05完整答案：把失败边界放在每个供应商周围

上一课E05原题是：broken场景、购买1件，boreal503时保留atlas与cedar，失败列表只记录boreal，最低价为cedar，并稳定排序。答案在lessons/06/answer_previous.py。它刻意继续使用串行for循环，先把“部分失败”的业务行为和“并发执行”的调度行为分开验收。

答案中的try只包住一次fetch；捕获ProviderFailure后追加一条失败，循环继续。成功报价统一交给summarize排序。若把整个循环包在一个try中，boreal失败之后cedar就没有机会被询问，即使最后仍能返回atlas，也不满足“尽量保留结果”的合同。末尾附答案全部源码。

从工作线程提交，到主线程领取

ThreadPoolExecutor(max_workers=3)最多同时运行三个工作线程。submit(tracked, supplier)提交可调用对象及参数，立即返回一个Future。Future是任务最终结果的位置，不是结果本身：它可能排队中、运行中、成功或失败。

用字典保存future -> supplier非常必要。请求失败时没有报价行可用于识别供应商，但Future仍能映射回提交对象。这和“函数返回了谁”是两条独立的信息渠道。主线程从as_completed(futures)取得已经完成的Future，再调用future.result()：成功时得到行，失败时在主线程重新抛出工作线程里的异常。

as_completed按完成情况交付，不承诺按提交顺序。正常延迟大约是cedar 0.09秒、atlas 0.12秒、boreal 0.18秒，所以三线程演示经常先观察到cedar，但操作系统调度、连接初始化和负载都能改变次序。程序打印completion_order帮助我们看这个事实，业务代码不能依赖某次观察到的顺序。

如果改成for future in futures: future.result()，请求依然已经并发提交，只是收集者可能卡在最早提交的慢任务上，迟迟无法处理后面早已完成的结果。它不等于完全串行，却失去了及时处理已完成工作的便利。这两个层次要分开：任务何时运行，由执行器决定；结果何时被主线程消费，由收集方式决定。

为什么收集列表不加锁，计数器却加锁

本课工作线程只执行work和完成计数，rows、failures、completion_order都由主线程修改，因此这些容器没有多线程写入，不需要给每个append包一把锁。减少共享状态通常比增加更多锁更容易推理。

completed刻意保留为共享状态，用来说明锁的边界。completed += 1在语义上包含读取旧值、计算新值、写回，不应把“某个CPython版本的一次压力测试没丢计数”当作原子性合同。GIL限制线程执行Python代码，不代表任意复合业务操作都自动不可分割，也不保证未来实现仍保持相同表现。

with lock:只围住计数更新，不围住HTTP调用。若持锁期间等待网络，其他线程想更新计数就只能等待同一把锁；若把整个请求也纳入同一个锁，三个线程又退化为一次一个请求。临界区越长，锁竞争越可能成为新的串行瓶颈。

计数更新放在finally，表示“已经结束的尝试数”，包括成功、503、校验失败。它不是“成功报价数”。正常quantity=2时，三个请求都完成，completed==3，但只有两个可售报价。指标名称必须对应实际语义，否则监控会把正常缺货误报为丢任务。

HTTPX同步Client可以在线程之间共享。本课只用它发送请求，不在工作线程中同时修改默认headers、cookies或其他配置。客户端在外层with创建，执行器退出并确认线程结束后才关闭它；如果先关闭客户端再让线程继续发请求，得到的会是生命周期错误，而不是供应商错误。

两个工作线程到底限制了什么

运行命令均从课程根目录执行，供应商保持在另一终端运行。Linux使用python3.11替换解释器名。

python tools/provider_lab.py--port 9101

python lessons/06/answer_previous.py python lessons/06/main.py--workers 3 python lessons/06/main.py--workers 2 python lessons/06/main.py--workers 2--scenario broken python lessons/06/timeout_demo.py python-m unittest discover-s lessons/06-p test_lesson.py-v

设成2以后，一开始最多两家运行，第三家在执行器队列中等待空位。max_workers约束正在执行的线程数，并不限制你总共调用多少次submit。这里总共只有三个任务，排队规模天然有限；如果把一百万个SKU都立即submit，线程仍然只有两个，但一百万个Future与参数仍要占内存。第10课会用有界队列控制生产速度。

本课测试使用Barrier让两个工作者必须碰面后才能继续。如果错误地改成串行执行，第一个工作者会等待不到第二个而失败；独立计数器还检查峰值恰为2。测试随后使用相同总价验证供应商名称排序。这个断言抓的是实际资源行为和可见结果，而不是只检查代码里是否写着max_workers=2。

最容易误用的超时：等不及不等于杀掉线程

timeout_demo.py用Event让线程停在一个可控位置。主线程调用future.result(timeout=0.02)，等待超时后打印running_after_timeout=True。紧接着调用future.cancel()，已经开始运行的任务返回False。只有仍在排队、还没有开始的任务才可能被cancel取消。

演示最后在finally里设置release事件，工作线程正常返回，避免留下一个永远阻塞的进程。这也说明线程池上下文管理器的退出语义：默认会等待工作者收尾。即使主线程已经捕获Future超时，离开with ThreadPoolExecutor(...)时仍可能继续等待。因此不能通过给result()设置0.8秒就宣称整个接口最多0.8秒。

真实网络请求需要HTTP库自身的连接、读取等超时；线程里的长CPU循环需要主动检查协作停止标记。如果底层第三方函数无法中止，线程池不能替你安全杀死它。把shutdown(wait=False)当成解决方案也不成立：那只让当前调用不等待，不会撤销已经运行的代码，解释器退出通常仍会等待线程池任务。

本课适合解释I/O重叠。对纯Python密集计算，常见有GIL的CPython中多个线程通常不能让Python字节码同时在多个核心上跑；下一课会为真正的CPU工作选择进程池。某些扩展库会释放GIL，因此“所有CPU代码用线程都无效”也过于武断，应该依据实际工作负载测量。

E06：把资源上限与业务顺序分开验收

原题E06：对三个正常供应商使用最多2个工作线程，购买2件。可售结果必须按(total_cents, supplier)稳定排序，正常结果依次是boreal 25900、atlas 26400，cedar被库存规则过滤。再构造atlas与boreal总价相同的数据，断言atlas在前；不能依赖Future完成顺序。用测试中的活动计数证明峰值不超过2，并让至少两个任务真实重叠。下一课提供完整答案。

遇到错误先判断所属层：若输出顺序偶尔变化，检查是否忘了最后排序；若峰值超过2，检查是否还有额外执行器；若超时后程序迟迟不退出，检查运行中的线程是否有自己的退出条件。并发程序的问题不总是锁的问题，找准被违反的合同更省力。

一把锁保护的是不变量，而不是某一行语法

设想我们同时维护“完成数量”和“总耗时”。如果它们必须描述同一批已完成任务，就应在同一个临界区内一起更新；只分别给两个赋值加锁，读取者可能看到新计数配旧耗时。设计锁之前，先用一句话写清必须始终成立的关系，再围住那段不可被别人看到中间状态的操作，比逐行到处加锁更可靠。

相反，不需要共享的东西尽量留在工作者局部变量里。每个请求的开始时刻、响应对象和解析中间值属于这次调用，不要为了“统一管理”放进全局字典让所有线程读写。主线程集中合并成功和失败，也是同一个设计思路：把跨线程协调缩到最少，保留清楚的数据流。

死锁往往来自不同代码以不同顺序获取多把锁。例如一条路径先拿计数锁再拿日志锁，另一条先拿日志锁再等计数锁，二者都可能握着对方需要的资源。当前例子只使用一把短时锁，避免过早引入这种组合。真实系统无法避免多锁时，应统一获取顺序，并避免持锁调用未知的外部回调或阻塞操作。

提交成功并不保证业务成功

submit返回Future，只表示执行器接受了任务，不表示HTTP已经发出，更不表示供应商成功。队列里的任务可能还没开始；任务执行后也可能返回技术失败。因此服务不能在提交后立即把completed加一，否则排队、运行和完成三个阶段会被混在一起。当前计数位于工作函数的finally，明确指向执行结束。

Future携带异常的行为也解释了为什么必须领取每个结果。如果只提交不调用result，某些业务异常可能长时间无人观察，表面上主线程没有报错，实际却丢了一家报价。as_completed循环不仅为了收集快结果，也是让每一个提交都有对应的结算位置。即使业务最终只选择最低价，其他供应商任务也不能变成无人负责的后台工作。

调线程数时先找瓶颈

把工作线程从一改成二，通常能重叠两段网络等待；继续改成几十却未必有收益。这里只有三项工作，多出来的额度没有可执行任务；请求量更大时，还会受客户端连接池、服务端容量、网络带宽以及本机调度限制。远端限流后增加线程，可能把成功吞吐换成更多快速失败。

公平比较应固定同一批输入、同样的失败策略和超时、同样的连接池生命周期，并在预热后重复多轮。只比较一次终端命令的壁钟时间，会把解释器启动、杀毒扫描和客户端初始化一起混进去。本课打印的是学习用观测值，第十七课才使用一致边界测量分位数；在那之前，能够证明峰值受限和结果不变已经比一个偶然漂亮的倍数更有说服力。

本课完整源码

domain.py与gateway.py是05课规则和同步网关的原样副本；独立保存是为了让本课目录不依赖未来项目。下面给出它们及所有新增实验的完整代码。

lessons/06/answer_previous.py

"""E05: HTTP 503 must preserve successful providers."""importosimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarizefromgatewayimportfetchdefsolve():rows,failures=[],[]base_url=os.getenv('QUOTEFLOW_PROVIDER_URL','http://127.0.0.1:9101')withhttpx.Client(trust_env=False)asclient:forsupplierinSUPPLIERS:try:rows.append(fetch(client,base_url,supplier,'broken',1))exceptProviderFailureasexc:failures.append({'supplier':supplier,'code':exc.code})result=summarize(rows,failures,1)assert{row['supplier']forrowinresult['quotes']}=={'atlas','cedar'}assertresult['best']['supplier']=='cedar'assertresult['failures']==[{'supplier':'boreal','code':'http_503'}]returnresultif__name__=='__main__':print(solve())

lessons/06/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/06/gateway.py

"""One synchronous HTTP attempt; no retry and no batch time budget."""fromtimeimportperf_counterimporthttpxfromdomainimportInvalidData,ProviderFailure,Quotedeffetch(client,base_url,supplier,scenario,quantity):"""只执行一次HTTP尝试；客户端归调用者所有，此处不创建、不关闭。"""started=perf_counter()try:response=client.get(f'{base_url}/quotes/{supplier}',params={'sku':'KB-001','scenario':scenario},timeout=httpx.Timeout(0.6))received=perf_counter()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')fromexc# HTTP成功不保证数据合法；解析错误在供应商边界转成明确类别。try:data=response.json()exceptValueErrorasexc:raiseInvalidData()fromexc quote=Quote.parse(data,supplier,'KB-001')row=quote.row(quantity)row['http_ms']=round((received-started)*1000,2)row['parse_ms']=round((perf_counter()-received)*1000,2)returnrow

lessons/06/main.py

"""Bounded thread pool: completion order differs from ranking order."""importargparsefromconcurrent.futuresimportThreadPoolExecutor,as_completedimportjsonimportosimportthreadingfromtimeimportperf_counterimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarize,validate_quantityfromgatewayimportfetchdefrun_pool(work,quantity=2,max_workers=3):"""限制运行线程数；结果只由主线程汇总，避免无必要的共享写入。"""validate_quantity(quantity)rows,failures,completion_order=[],[],[]completed=0lock=threading.Lock()deftracked(supplier):# 计数代表结束的尝试，包含失败；锁只保护复合更新，不包住HTTP等待。nonlocalcompletedtry:returnwork(supplier)finally:withlock:completed+=1withThreadPoolExecutor(max_workers=max_workers)aspool:futures={pool.submit(tracked,supplier):supplierforsupplierinSUPPLIERS}forfutureinas_completed(futures):supplier=futures[future]completion_order.append(supplier)try:rows.append(future.result())exceptProviderFailureasexc:failures.append({'supplier':supplier,'code':exc.code})result=summarize(rows,failures,quantity)result.update(completed=completed,completion_order=completion_order)returnresultdefmain():parser=argparse.ArgumentParser()parser.add_argument('--scenario',default='normal')parser.add_argument('--workers',type=int,default=3)args=parser.parse_args()base_url=os.getenv('QUOTEFLOW_PROVIDER_URL','http://127.0.0.1:9101')started=perf_counter()withhttpx.Client(trust_env=False)asclient:result=run_pool(lambdaname:fetch(client,base_url,name,args.scenario,2),max_workers=args.workers)result['elapsed_ms']=round((perf_counter()-started)*1000,2)print(json.dumps(result,ensure_ascii=False,indent=2))if__name__=='__main__':main()

lessons/06/test_lesson.py

importthreadingimporttimeimportunittestfrommainimportrun_poolclassLessonTests(unittest.TestCase):deftest_two_threads_overlap_but_never_exceed_capacity_and_sort(self):lock=threading.Lock()active=peak=0ready=threading.Barrier(2)defwork(name):nonlocalactive,peakwithlock:active+=1peak=max(peak,active)try:ifnamein('atlas','boreal'):ready.wait(timeout=2)time.sleep(0.02)returndict(supplier=name,total_cents=100,stock=100)finally:withlock:active-=1result=run_pool(work,quantity=2,max_workers=2)self.assertEqual(peak,2)self.assertEqual(active,0)self.assertEqual([q['supplier']forqinresult['quotes']],['atlas','boreal','cedar'])self.assertEqual(result['completed'],3)if__name__=='__main__':unittest.main()

lessons/06/timeout_demo.py

"""A future timeout only stops waiting; it cannot kill a running thread."""fromconcurrent.futuresimportThreadPoolExecutor,TimeoutErrorfromthreadingimportEventdefmain():entered,release=Event(),Event()defwork():entered.set()release.wait()return'finished after release'withThreadPoolExecutor(max_workers=1)aspool:future=pool.submit(work)try:assertentered.wait(timeout=2)try:future.result(timeout=0.02)exceptTimeoutError:print({'running_after_timeout':future.running(),'cancel_succeeded':future.cancel()})else:raiseAssertionError('worker should still be waiting')assertfuture.running()finally:release.set()print(future.result(timeout=2))if__name__=='__main__':main()