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

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_workers3)最多同时运行三个工作线程。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、校验失败。它不是“成功报价数”。正常quantity2时三个请求都完成completed3但只有两个可售报价。指标名称必须对应实际语义否则监控会把正常缺货误报为丢任务。HTTPX同步Client可以在线程之间共享。本课只用它发送请求不在工作线程中同时修改默认headers、cookies或其他配置。客户端在外层with创建执行器退出并确认线程结束后才关闭它如果先关闭客户端再让线程继续发请求得到的会是生命周期错误而不是供应商错误。两个工作线程到底限制了什么运行命令均从课程根目录执行供应商保持在另一终端运行。Linux使用python3.11替换解释器名。python tools/provider_lab.py--port 9101python 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_workers2。最容易误用的超时等不及不等于杀掉线程timeout_demo.py用Event让线程停在一个可控位置。主线程调用future.result(timeout0.02)等待超时后打印running_after_timeoutTrue。紧接着调用future.cancel()已经开始运行的任务返回False。只有仍在排队、还没有开始的任务才可能被cancel取消。演示最后在finally里设置release事件工作线程正常返回避免留下一个永远阻塞的进程。这也说明线程池上下文管理器的退出语义默认会等待工作者收尾。即使主线程已经捕获Future超时离开with ThreadPoolExecutor(...)时仍可能继续等待。因此不能通过给result()设置0.8秒就宣称整个接口最多0.8秒。真实网络请求需要HTTP库自身的连接、读取等超时线程里的长CPU循环需要主动检查协作停止标记。如果底层第三方函数无法中止线程池不能替你安全杀死它。把shutdown(waitFalse)当成解决方案也不成立那只让当前调用不等待不会撤销已经运行的代码解释器退出通常仍会等待线程池任务。本课适合解释I/O重叠。对纯Python密集计算常见有GIL的CPython中多个线程通常不能让Python字节码同时在多个核心上跑下一课会为真正的CPU工作选择进程池。某些扩展库会释放GIL因此“所有CPU代码用线程都无效”也过于武断应该依据实际工作负载测量。E06把资源上限与业务顺序分开验收原题E06对三个正常供应商使用最多2个工作线程购买2件。可售结果必须按(total_cents, supplier)稳定排序正常结果依次是boreal 25900、atlas 26400cedar被库存规则过滤。再构造atlas与boreal总价相同的数据断言atlas在前不能依赖Future完成顺序。用测试中的活动计数证明峰值不超过2并让至少两个任务真实重叠。下一课提供完整答案。遇到错误先判断所属层若输出顺序偶尔变化检查是否忘了最后排序若峰值超过2检查是否还有额外执行器若超时后程序迟迟不退出检查运行中的线程是否有自己的退出条件。并发程序的问题不总是锁的问题找准被违反的合同更省力。一把锁保护的是不变量而不是某一行语法设想我们同时维护“完成数量”和“总耗时”。如果它们必须描述同一批已完成任务就应在同一个临界区内一起更新只分别给两个赋值加锁读取者可能看到新计数配旧耗时。设计锁之前先用一句话写清必须始终成立的关系再围住那段不可被别人看到中间状态的操作比逐行到处加锁更可靠。相反不需要共享的东西尽量留在工作者局部变量里。每个请求的开始时刻、响应对象和解析中间值属于这次调用不要为了“统一管理”放进全局字典让所有线程读写。主线程集中合并成功和失败也是同一个设计思路把跨线程协调缩到最少保留清楚的数据流。死锁往往来自不同代码以不同顺序获取多把锁。例如一条路径先拿计数锁再拿日志锁另一条先拿日志锁再等计数锁二者都可能握着对方需要的资源。当前例子只使用一把短时锁避免过早引入这种组合。真实系统无法避免多锁时应统一获取顺序并避免持锁调用未知的外部回调或阻塞操作。提交成功并不保证业务成功submit返回Future只表示执行器接受了任务不表示HTTP已经发出更不表示供应商成功。队列里的任务可能还没开始任务执行后也可能返回技术失败。因此服务不能在提交后立即把completed加一否则排队、运行和完成三个阶段会被混在一起。当前计数位于工作函数的finally明确指向执行结束。Future携带异常的行为也解释了为什么必须领取每个结果。如果只提交不调用result某些业务异常可能长时间无人观察表面上主线程没有报错实际却丢了一家报价。as_completed循环不仅为了收集快结果也是让每一个提交都有对应的结算位置。即使业务最终只选择最低价其他供应商任务也不能变成无人负责的后台工作。调线程数时先找瓶颈把工作线程从一改成二通常能重叠两段网络等待继续改成几十却未必有收益。这里只有三项工作多出来的额度没有可执行任务请求量更大时还会受客户端连接池、服务端容量、网络带宽以及本机调度限制。远端限流后增加线程可能把成功吞吐换成更多快速失败。公平比较应固定同一批输入、同样的失败策略和超时、同样的连接池生命周期并在预热后重复多轮。只比较一次终端命令的壁钟时间会把解释器启动、杀毒扫描和客户端初始化一起混进去。本课打印的是学习用观测值第十七课才使用一致边界测量分位数在那之前能够证明峰值受限和结果不变已经比一个偶然漂亮的倍数更有说服力。本课完整源码domain.py与gateway.py是05课规则和同步网关的原样副本独立保存是为了让本课目录不依赖未来项目。下面给出它们及所有新增实验的完整代码。lessons/06/answer_previous.pyE05: HTTP 503 must preserve successful providers.importosimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarizefromgatewayimportfetchdefsolve():rows,failures[],[]base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101)withhttpx.Client(trust_envFalse)asclient:forsupplierinSUPPLIERS:try:rows.append(fetch(client,base_url,supplier,broken,1))exceptProviderFailureasexc:failures.append({supplier:supplier,code:exc.code})resultsummarize(rows,failures,1)assert{row[supplier]forrowinresult[quotes]}{atlas,cedar}assertresult[best][supplier]cedarassertresult[failures][{supplier:boreal,code:http_503}]returnresultif__name____main__:print(solve())lessons/06/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/06/gateway.pyOne synchronous HTTP attempt; no retry and no batch time budget.fromtimeimportperf_counterimporthttpxfromdomainimportInvalidData,ProviderFailure,Quotedeffetch(client,base_url,supplier,scenario,quantity):只执行一次HTTP尝试客户端归调用者所有此处不创建、不关闭。startedperf_counter()try:responseclient.get(f{base_url}/quotes/{supplier},params{sku:KB-001,scenario:scenario},timeouthttpx.Timeout(0.6))receivedperf_counter()response.raise_for_status()excepthttpx.TimeoutExceptionasexc:raiseProviderFailure(timeout)fromexcexcepthttpx.HTTPStatusErrorasexc:raiseProviderFailure(fhttp_{exc.response.status_code})fromexcexcepthttpx.DecodingErrorasexc:# 坏gzip等内容编码错误是远端数据损坏不归类为可重试传输故障。raiseInvalidData()fromexcexcepthttpx.RequestErrorasexc:raiseProviderFailure(transport)fromexc# HTTP成功不保证数据合法解析错误在供应商边界转成明确类别。try:dataresponse.json()exceptValueErrorasexc:raiseInvalidData()fromexc quoteQuote.parse(data,supplier,KB-001)rowquote.row(quantity)row[http_ms]round((received-started)*1000,2)row[parse_ms]round((perf_counter()-received)*1000,2)returnrowlessons/06/main.pyBounded thread pool: completion order differs from ranking order.importargparsefromconcurrent.futuresimportThreadPoolExecutor,as_completedimportjsonimportosimportthreadingfromtimeimportperf_counterimporthttpxfromdomainimportSUPPLIERS,ProviderFailure,summarize,validate_quantityfromgatewayimportfetchdefrun_pool(work,quantity2,max_workers3):限制运行线程数结果只由主线程汇总避免无必要的共享写入。validate_quantity(quantity)rows,failures,completion_order[],[],[]completed0lockthreading.Lock()deftracked(supplier):# 计数代表结束的尝试包含失败锁只保护复合更新不包住HTTP等待。nonlocalcompletedtry:returnwork(supplier)finally:withlock:completed1withThreadPoolExecutor(max_workersmax_workers)aspool:futures{pool.submit(tracked,supplier):supplierforsupplierinSUPPLIERS}forfutureinas_completed(futures):supplierfutures[future]completion_order.append(supplier)try:rows.append(future.result())exceptProviderFailureasexc:failures.append({supplier:supplier,code:exc.code})resultsummarize(rows,failures,quantity)result.update(completedcompleted,completion_ordercompletion_order)returnresultdefmain():parserargparse.ArgumentParser()parser.add_argument(--scenario,defaultnormal)parser.add_argument(--workers,typeint,default3)argsparser.parse_args()base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101)startedperf_counter()withhttpx.Client(trust_envFalse)asclient:resultrun_pool(lambdaname:fetch(client,base_url,name,args.scenario,2),max_workersargs.workers)result[elapsed_ms]round((perf_counter()-started)*1000,2)print(json.dumps(result,ensure_asciiFalse,indent2))if__name____main__:main()lessons/06/test_lesson.pyimportthreadingimporttimeimportunittestfrommainimportrun_poolclassLessonTests(unittest.TestCase):deftest_two_threads_overlap_but_never_exceed_capacity_and_sort(self):lockthreading.Lock()activepeak0readythreading.Barrier(2)defwork(name):nonlocalactive,peakwithlock:active1peakmax(peak,active)try:ifnamein(atlas,boreal):ready.wait(timeout2)time.sleep(0.02)returndict(suppliername,total_cents100,stock100)finally:withlock:active-1resultrun_pool(work,quantity2,max_workers2)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.pyA future timeout only stops waiting; it cannot kill a running thread.fromconcurrent.futuresimportThreadPoolExecutor,TimeoutErrorfromthreadingimportEventdefmain():entered,releaseEvent(),Event()defwork():entered.set()release.wait()returnfinished after releasewithThreadPoolExecutor(max_workers1)aspool:futurepool.submit(work)try:assertentered.wait(timeout2)try:future.result(timeout0.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(timeout2))if__name____main__:main()