Python进阶与并发编程13：八个用户同时查询，为什么每个人的Semaphore都管不住总量

Python进阶与并发编程13八个用户同时查询为什么每个人的Semaphore都管不住总量上一课一次请求只发三个供应商调用速度似乎已经很好。现在同时来八个请求就可能同时制造二十四个出站操作。若在路由内部写Semaphore(6)每个请求都有自己的六个槽位三个供应商甚至用不满它应用总量仍是二十四。问题不在Semaphore失灵而在资源所有权放错了地方想约束所有请求就必须让所有请求竞争同一个对象。本课在checkpoints/13实现每进程出站6、入站8、每次查询总预算0.8秒。尚未加入重试与批量接口。先建立稳定边界下一课才能讨论失败重试会额外消耗多少资源。E12答案422是一条没有网络副作用的路径完整答案为lessons/13/answer_previous.py从课程根目录运行python lessons/13/answer_previous.py。脚本明确把第12课阶段源码放入导入路径因此即使最终项目已安装也会验证上课的代码。MockTransport的处理器一旦被访问就增加计数ASGITransport提交quantity0断言422之后再断言计数为0。这个断言捕捉的是有实际意义的错误如果有人把校验移到报价逻辑之后HTTP响应可能仍是422供应商却已被访问。只断言状态码会漏掉资源浪费。测试显式进入app.router.lifespan_context(app)因为httpx的ASGITransport不会自动帮我们执行应用启动和关闭。测试中借用的客户端与线上lifespan使用同一构造路径只在HTTP传输边界替换实现。数量严格校验也让后面的容量讨论有清晰前提无效输入不参与准入计数不产生三家调用。有效输入即使最终没有报价也是一条真实的业务请求需要预算与容量保护。三种上限分别回答三个问题入站8回答“同时允许多少个业务请求留在应用中”出站6回答“同时允许多少次供应商HTTP尝试”连接池6回答“客户端最多管理多少条HTTP连接”。三者相关却不能互相替代。只限制连接数大量已接受的请求仍可以在池内排队持有任务、请求体和响应上下文。只限制业务请求数单个请求也可能派生很多出站任务。只限制出站信号量没有连接池上限则底层资源配置与业务预期缺少一致性。当前每个入站请求有三个固定供应商任务八个请求最多创建二十四个这样的孩子。其中最多六个持有出站信号量其余在等待。任务数量已经被入站上限和固定供应商集合共同约束。信号量是等待机制入站准入则选择立即拒绝第九个请求收到503与Retry-After不进入一个没有边界的新队列。立即拒绝不是说明第九位客户没有价值而是明确过载时的合同。让所有请求无期限等待经常把负载问题伪装成更长的尾延迟有限的响应时间必须同时考虑能接受多少等待者。此处HTTP503表示服务当前容量不足不与单个供应商返回的503混淆前者是接口状态后者出现在HTTP200响应的failures中。准入计数为什么没有锁admission是同步上下文管理器检查active_requests inbound_limit未满则加一退出时在finally减一。检查和加一之间没有await且这个服务实例只由同一事件循环中的协程使用所以其他任务不会在两行之间获得运行机会。它与第六课线程共享计数时必须加锁并不矛盾。这不是“Python有GIL所以两行代码天然原子”。GIL是解释器执行线程的机制一段业务判断与递增跨线程仍可能交错。这里依靠的是事件循环任务在显式挂起点合作调度、关键区内部没有挂起点的事实。如果以后在线程里直接修改这个对象或者把准入判断改成await调用外部存储就必须重新审视同步机制。注释应该说明约束条件不能把当前成立的推理扩大成语言保证。with admission()包住整个查询成功、超时以及外部取消都会离开上下文并归还入站槽位。Busy发生在计数递增之前因此被拒绝的请求没有拿到资源也不应再执行减一。这是E13会验证的重点拒绝一条请求不能使容量被消耗也不能把计数减成负数。构造器拒绝非正预算、非正入站和出站上限。把Semaphore设为0可以合法构造但永远没有初始槽位可取学习者调整参数时看到明确ValueError比看到整个服务安静地等待更容易定位问题。配置错误应该尽早暴露而不是等到用户请求承受超时才显现。超时值不是总预算截止时间才容易组合HTTP客户端的timeout描述网络操作阶段的等待限制并不自动涵盖应用自己的Semaphore排队。假设一个调用先排队0.6秒再获得连接执行0.7秒每个局部步骤都可能未超过0.8秒用户却等了1.3秒。若后面再加重试这种“每一步都合法整体严重超时”会更明显。我们在查询服务开始时记录deadline loop.time() 0.8随后用asyncio.timeout_at(deadline)包住TaskGroup。所有供应商任务共享同一绝对截止时刻。它们即使创建时间不同、排队长度不同剩余预算都来自同一个时钟没有谁能在拿到槽位之后重新获得完整0.8秒。loop.time()是事件循环使用的单调时钟适合计算截止点不要用会受系统校时影响的墙钟做这种差值。用于展示的elapsed_ms来自perf_counter只负责测量经过时间。内部绝对截止点和对外耗时是两个用途不能把前者直接序列化为客户看得懂的时间戳。这里预算的起点是请求校验后进入业务服务覆盖业务排队、HTTP等待及后续章节的退避不包含客户端上传请求体、响应序列化和响应发送。有限预算是一项具体合同不能把它宣传成从客户端按下按钮到接收最后一个字节的端到端保证。事件循环若被同步代码阻塞协作式取消也无法在精确第800毫秒执行因此测试用合理宽容度检查边界而非追求毫秒级实时系统精度。超时后为什么还能保留atlas每个供应商任务拿到结果后立即写入completed[supplier]。到截止时间timeout_at取消当前收集任务TaskGroup接着取消并等待尚未完成的孩子已写入的成功结果仍在completed中。等收尾结束_result按固定供应商列表构造响应已有成功且库存够的报价入榜已有Failure保留未完成者记timeout。这与“用gather等全部结果外面超时后丢弃整个列表”不同客户端仍能拿到截止前完成的有用工作。正常slow场景下boreal睡1.5秒atlas通常早已结束因此响应可保留atlas并把boreal标为timeout。缺货的cedar依旧只是过滤不能因为未出现在quotes里便自动补一个失败。代码只在timeout上下文外捕获TimeoutError。由timeout_at自己发起的取消会在那里转换为TimeoutError调用方主动cancel产生的CancelledError继续向上传播。若粗暴捕获BaseException或者把CancelledError变成普通Failure调用方就失去取消整条工作链的能力服务器关闭或上游超时也可能留下继续运行的任务。信号量使用async with因此在HTTP等待期间取消会释放槽位TaskGroup退出等待子任务收尾所以方法返回时不会留下由这个查询创建、仍在后台占用本地槽位的孩子。但是已经发出的真实HTTP请求可能仍在供应商服务器上执行关闭本地连接不等于杀掉远端线程。测量供应商active时必须牢记这两个观察位置的差别。失败退出也要做相同的归还验证。测试让boreal返回带gzip头的坏压缩流HTTPX在client.get内部解码失败单独捕获DecodingError之后结果应为invalid_dataHTTP接口仍200atlas和cedar保留。测试跟踪响应流关闭并在结束后通过公开acquire方法重新取得全部六个出站槽位随后逐一release这样能发现只泄漏一个槽位、但下一次单请求碰巧还能运行的缺陷。用可控测试证明共享、排队和取消本阶段测试使用MockTransport刻意让每个出站调用短暂await记录同时active的峰值。并行发起八次query后峰值应为6这能发现“误把信号量放进每次query”的回归。另一个测试把出站设为1、预算缩至0.05秒、供应商延迟设为2秒断言连正在等槽位的供应商也被总预算覆盖并且结束后active与入站计数均为0。取消测试使用Event建立确定的执行顺序先确认任务进入供应商再cancel并await退出最后复用同一个服务实例成功查询。观察“下一次能运行”比访问Semaphore的私有_value更贴近业务需要。真正的程序错误则应该通过ExceptionGroup暴露而不是被归入timeout。真实HTTP压力测试需要两台本地进程——供应商与QuoteFlow。normal场景下可以用供应商的max_active检验观测峰值slow取消后供应商可能仍未结束因此必须等active归零再reset统计避免把前一轮的远端余留工作算到下一轮。MockTransport没有TCP连接也不能验证连接复用、监听端口或网络超时两类测试分别提供证据不能互相冒充。安装和启动方式仍是进入checkpoints/13建立独立Python3.11环境执行python -m pip install -e .[test]、python -m pytest -q与python -m uvicorn quoteflow.api:app --host 127.0.0.1 --port 9100。供应商从课程根目录启动python tools/provider_lab.py --port 9101。Windows用该阶段.venv\Scripts\python.exeLinux用.venv/bin/python不要借用一个安装着其他阶段的环境。看待一次超时需要同时看服务与供应商把slow请求返回的timeout理解成“我在约定时间内没有拿到这家结果”比理解成“远端已经停止工作”准确。我们能取消自己的等待不能沿着TCP连接倒过来销毁供应商内部的业务线程。若供应商执行昂贵查询它需要自己的取消协议或预算传播机制服务端才能真正减少已经发出的工作本课实验没有实现那个额外协议。因此容量测试先用normal场景测共享出站峰值再单独测slow预算和本地资源归还。每种测试只宣称它看到的边界避免一个漂亮的max_active数字掩盖远端的残留工作。这个习惯以后也适用于数据库、消息队列与云API取消客户端协程从来不是对所有依赖系统状态的最终裁决。E13拒绝之后这个服务还能接下一位客户吗用inbound_limit1创建服务让第一条请求停在Event上第二条必须抛Busy。放行第一条并等待它完成断言active_requests回到0再发第三条并成功。HTTP层另断言Busy映射503Retry-After为1供应商失败仍应以200响应里的failures呈现。完整答案将在第14课的lessons/14/answer_previous.py。这些上限都是每进程的。若用四个worker通常就有四个客户端池、四份入站计数和四个出站信号量总容量会随进程数放大。跨进程、跨机器的配额需要另一个协调层本课不假装一个本地Semaphore能承担那个职责。客户端断开也不保证ASGI框架自动取消handler我们当前验证的是显式任务取消路径不能据此声称所有断连都能立即回收计算。本课改动的模型、服务与入口全文附后模型虽未改变也完整列出方便单独重建阶段。你可以沿着“谁拥有资源、谁等待资源、谁归还资源”的顺序阅读而无需记一张抽象术语表。本课完整代码lessons/13/answer_previous.pyE12 答案422 必须发生在供应商访问之前直接使用第12课阶段。importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/checkpoints/12/src))fromquoteflow.apiimportcreate_appasyncdefmain():calls0asyncdefsupplier(request):nonlocalcalls calls1returnhttpx.Response(503)appcreate_app(transporthttpx.MockTransport(supplier))asyncwithapp.router.lifespan_context(app):asyncwithhttpx.AsyncClient(transporthttpx.ASGITransport(appapp),base_urlhttp://api)asclient:responseawaitclient.post(/api/quotes,json{quantity:0})assertresponse.status_code422assertcalls0print(E12 passed: quantity0 returns 422; supplier calls0)if__name____main__:asyncio.run(main())checkpoints/13/pyproject.toml[build-system] requires [setuptools68,81] build-backend setuptools.build_meta [project] name quoteflow-course version 1.0.0 description Python 3.11 bounded concurrency teaching service requires-python 3.11 dependencies [fastapi0.115.12, httpx0.28.1, uvicorn0.34.2, pydantic2.7,3] [project.optional-dependencies] test [pytest8.3.5] [tool.pytest.ini_options] testpaths [tests] pythonpath [src]checkpoints/13/src/quoteflow/__init__.pyStandalone course checkpoint.checkpoints/13/src/quoteflow/models.pyfromtypingimportAnnotated,LiteralfrompydanticimportBaseModel,ConfigDict,Field SupplierLiteral[atlas,boreal,cedar]ScenarioLiteral[normal,slow,broken,malformed,flaky]NonnegativeIntAnnotated[int,Field(strictTrue,ge0)]classQuoteRequest(BaseModel):入口严格校验布尔值、字符串数量和调用方自定义 URL 均被拒绝。model_configConfigDict(extraforbid)sku:strField(defaultKB-001,min_length1,max_length64)quantity:intField(default2,strictTrue,ge1,le100)scenario:ScenarionormalclassSupplierQuote(BaseModel):供应商是另一处不可信边界金额分与库存必须为非负整数。model_configConfigDict(extraforbid,frozenTrue)supplier:Supplier sku:strunit_cents:NonnegativeInt shipping_cents:NonnegativeInt stock:NonnegativeIntclassRankedQuote(SupplierQuote):total_cents:NonnegativeIntclassFailure(BaseModel):supplier:Supplier code:strclassQuoteResult(BaseModel):sku:strquantity:intquotes:list[RankedQuote]failures:list[Failure]best:RankedQuote|Noneelapsed_ms:floatrequest_id:strcheckpoints/13/src/quoteflow/service.py每进程一个服务实例所有 HTTP handler 共用准入计数与出站信号量。importasynciofromcontextlibimportcontextmanagerfromtimeimportperf_counterfromuuidimportuuid4importhttpxfrompydanticimportValidationErrorfrom.modelsimportFailure,QuoteRequest,QuoteResult,RankedQuote,SupplierQuote SUPPLIERS(atlas,boreal,cedar)classBusy(Exception):入站容量已满入口把它翻译为 HTTP 503。classQuoteService:def__init__(self,client:httpx.AsyncClient,*,budget:float0.8,inbound_limit:int8,outbound_limit:int6):ifbudget0orinbound_limit0oroutbound_limit0:raiseValueError(budget and concurrency limits must be positive)self.clientclient self.budgetbudget self.inbound_limitinbound_limit self.active_requests0self.outboundasyncio.Semaphore(outbound_limit)contextmanagerdefadmission(self):# 检查和递增之间没有 await同一事件循环的任务不会在此交错。# 这不是跨线程保证更不是依靠 GIL 实现的业务原子性。ifself.active_requestsself.inbound_limit:raiseBusy(inbound capacity exhausted)self.active_requests1try:yieldfinally:self.active_requests-1asyncdef_fetch(self,supplier:str,query:QuoteRequest,request_id:str):try:asyncwithself.outbound:responseawaitself.client.get(f/quotes/{supplier},params{sku:query.sku,scenario:query.scenario},headers{X-Request-ID:request_id},)excepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生属于供应商数据错误。returnFailure(suppliersupplier,codeinvalid_data)excepthttpx.TransportError:returnFailure(suppliersupplier,codetransport_error)ifresponse.status_code!200:returnFailure(suppliersupplier,codefhttp_{response.status_code})try:quoteSupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(suppliersupplier,codeinvalid_data)ifquote.supplier!supplierorquote.sku!query.sku:returnFailure(suppliersupplier,codeinvalid_data)returnquoteasyncdef_collect(self,query:QuoteRequest,deadline:float,started:float,request_id:str)-QuoteResult:completed{}asyncdefone(supplier):completed[supplier]awaitself._fetch(supplier,query,request_id)# 预算覆盖信号量排队与本次 HTTP 尝试所有供应商共享截止时刻。# timeout_at 只把它发起的取消转换为 TimeoutError。# 外部 CancelledError 继续传播TaskGroup 先收尾所有子任务。try:asyncwithasyncio.timeout_at(deadline):asyncwithasyncio.TaskGroup()asgroup:forsupplierinSUPPLIERS:group.create_task(one(supplier))exceptTimeoutError:passreturnself._result(query,completed,started,request_id)staticmethoddef_result(query,completed,started,request_id):quotes,failures[],[]forsupplierinSUPPLIERS:valuecompleted.get(supplier)ifvalueisNone:failures.append(Failure(suppliersupplier,codetimeout))elifisinstance(value,Failure):failures.append(value)elifvalue.stockquery.quantity:quotes.append(RankedQuote(**value.model_dump(),total_cents(value.unit_cents*query.quantityvalue.shipping_cents)))quotes.sort(keylambdaquote:(quote.total_cents,quote.supplier))returnQuoteResult(skuquery.sku,quantityquery.quantity,quotesquotes,failuresfailures,bestquotes[0]ifquoteselseNone,elapsed_msround((perf_counter()-started)*1000,3),request_idrequest_id)asyncdefquote(self,query:QuoteRequest)-QuoteResult:startedperf_counter()deadlineasyncio.get_running_loop().time()self.budgetwithself.admission():returnawaitself._collect(query,deadline,started,uuid4().hex)checkpoints/13/src/quoteflow/api.pyimportosfromcontextlibimportasynccontextmanagerimporthttpxfromfastapiimportFastAPI,HTTPException,Requestfrom.modelsimportQuoteRequest,QuoteResultfrom.serviceimportBusy,QuoteServicedefcreate_app(*,transportNone)-FastAPI:asynccontextmanagerasyncdeflifespan(app):# 启动时创建一次应用关闭时等待 aclose每次请求只借用客户端。asyncwithhttpx.AsyncClient(base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101),timeouthttpx.Timeout(0.8),limitshttpx.Limits(max_connections6,max_keepalive_connections6),transporttransport,trust_envFalse,follow_redirectsFalse,)asclient:app.state.serviceQuoteService(client)yieldappFastAPI(titleQuoteFlow,version1.0.0,lifespanlifespan)app.get(/health)asyncdefhealth():return{status:ok}app.post(/api/quotes,response_modelQuoteResult)asyncdefquotes(query:QuoteRequest,request:Request):try:returnawaitrequest.app.state.service.quote(query)exceptBusyasexc:raiseHTTPException(503,capacity_exhausted,headers{Retry-After:1})fromexcreturnapp appcreate_app()