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提交quantity=0；断言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_data，HTTP接口仍200，atlas和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.exe，Linux用.venv/bin/python，不要借用一个安装着其他阶段的环境。
看待一次超时，需要同时看服务与供应商
把slow请求返回的timeout理解成“我在约定时间内没有拿到这家结果”，比理解成“远端已经停止工作”准确。我们能取消自己的等待，不能沿着TCP连接倒过来销毁供应商内部的业务线程。若供应商执行昂贵查询，它需要自己的取消协议或预算传播机制，服务端才能真正减少已经发出的工作；本课实验没有实现那个额外协议。
因此容量测试先用normal场景测共享出站峰值，再单独测slow预算和本地资源归还。每种测试只宣称它看到的边界，避免一个漂亮的max_active数字掩盖远端的残留工作。这个习惯以后也适用于数据库、消息队列与云API：取消客户端协程，从来不是对所有依赖系统状态的最终裁决。
E13：拒绝之后，这个服务还能接下一位客户吗
用inbound_limit=1创建服务，让第一条请求停在Event上；第二条必须抛Busy。放行第一条并等待它完成，断言active_requests回到0，再发第三条并成功。HTTP层另断言Busy映射503，Retry-After为1；供应商失败仍应以200响应里的failures呈现。完整答案将在第14课的lessons/14/answer_previous.py。
这些上限都是每进程的。若用四个worker，通常就有四个客户端池、四份入站计数和四个出站信号量，总容量会随进程数放大。跨进程、跨机器的配额需要另一个协调层，本课不假装一个本地Semaphore能承担那个职责。客户端断开也不保证ASGI框架自动取消handler；我们当前验证的是显式任务取消路径，不能据此声称所有断连都能立即回收计算。
本课改动的模型、服务与入口全文附后；模型虽未改变也完整列出，方便单独重建阶段。你可以沿着“谁拥有资源、谁等待资源、谁归还资源”的顺序阅读，而无需记一张抽象术语表。
本课完整代码
lessons/13/answer_previous.py
"""E12 答案：422 必须发生在供应商访问之前，直接使用第12课阶段。"""importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/"checkpoints/12/src"))fromquoteflow.apiimportcreate_appasyncdefmain():calls=0asyncdefsupplier(request):nonlocalcalls calls+=1returnhttpx.Response(503)app=create_app(transport=httpx.MockTransport(supplier))asyncwithapp.router.lifespan_context(app):asyncwithhttpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url="http://api")asclient:response=awaitclient.post("/api/quotes",json={"quantity":0})assertresponse.status_code==422assertcalls==0print("E12 passed: quantity=0 returns 422; supplier calls=0")if__name__=="__main__":asyncio.run(main())checkpoints/13/pyproject.toml
[build-system] requires = ["setuptools>=68,<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 = ["fastapi==0.115.12", "httpx==0.28.1", "uvicorn==0.34.2", "pydantic>=2.7,<3"] [project.optional-dependencies] test = ["pytest==8.3.5"] [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["src"]checkpoints/13/src/quoteflow/__init__.py
"""Standalone course checkpoint."""checkpoints/13/src/quoteflow/models.py
fromtypingimportAnnotated,LiteralfrompydanticimportBaseModel,ConfigDict,Field Supplier=Literal["atlas","boreal","cedar"]Scenario=Literal["normal","slow","broken","malformed","flaky"]NonnegativeInt=Annotated[int,Field(strict=True,ge=0)]classQuoteRequest(BaseModel):"""入口严格校验；布尔值、字符串数量和调用方自定义 URL 均被拒绝。"""model_config=ConfigDict(extra="forbid")sku:str=Field(default="KB-001",min_length=1,max_length=64)quantity:int=Field(default=2,strict=True,ge=1,le=100)scenario:Scenario="normal"classSupplierQuote(BaseModel):"""供应商是另一处不可信边界：金额分与库存必须为非负整数。"""model_config=ConfigDict(extra="forbid",frozen=True)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:float=0.8,inbound_limit:int=8,outbound_limit:int=6):ifbudget<=0orinbound_limit<=0oroutbound_limit<=0:raiseValueError("budget and concurrency limits must be positive")self.client=client self.budget=budget self.inbound_limit=inbound_limit self.active_requests=0self.outbound=asyncio.Semaphore(outbound_limit)@contextmanagerdefadmission(self):# 检查和递增之间没有 await：同一事件循环的任务不会在此交错。# 这不是跨线程保证，更不是依靠 GIL 实现的业务原子性。ifself.active_requests>=self.inbound_limit:raiseBusy("inbound capacity exhausted")self.active_requests+=1try:yieldfinally:self.active_requests-=1asyncdef_fetch(self,supplier:str,query:QuoteRequest,request_id:str):try:asyncwithself.outbound:response=awaitself.client.get(f"/quotes/{supplier}",params={"sku":query.sku,"scenario":query.scenario},headers={"X-Request-ID":request_id},)excepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生，属于供应商数据错误。returnFailure(supplier=supplier,code="invalid_data")excepthttpx.TransportError:returnFailure(supplier=supplier,code="transport_error")ifresponse.status_code!=200:returnFailure(supplier=supplier,code=f"http_{response.status_code}")try:quote=SupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(supplier=supplier,code="invalid_data")ifquote.supplier!=supplierorquote.sku!=query.sku:returnFailure(supplier=supplier,code="invalid_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:value=completed.get(supplier)ifvalueisNone:failures.append(Failure(supplier=supplier,code="timeout"))elifisinstance(value,Failure):failures.append(value)elifvalue.stock>=query.quantity:quotes.append(RankedQuote(**value.model_dump(),total_cents=(value.unit_cents*query.quantity+value.shipping_cents)))quotes.sort(key=lambdaquote:(quote.total_cents,quote.supplier))returnQuoteResult(sku=query.sku,quantity=query.quantity,quotes=quotes,failures=failures,best=quotes[0]ifquoteselseNone,elapsed_ms=round((perf_counter()-started)*1000,3),request_id=request_id)asyncdefquote(self,query:QuoteRequest)->QuoteResult:started=perf_counter()deadline=asyncio.get_running_loop().time()+self.budgetwithself.admission():returnawaitself._collect(query,deadline,started,uuid4().hex)checkpoints/13/src/quoteflow/api.py
importosfromcontextlibimportasynccontextmanagerimporthttpxfromfastapiimportFastAPI,HTTPException,Requestfrom.modelsimportQuoteRequest,QuoteResultfrom.serviceimportBusy,QuoteServicedefcreate_app(*,transport=None)->FastAPI:@asynccontextmanagerasyncdeflifespan(app):# 启动时创建一次，应用关闭时等待 aclose；每次请求只借用客户端。asyncwithhttpx.AsyncClient(base_url=os.getenv("QUOTEFLOW_PROVIDER_URL","http://127.0.0.1:9101"),timeout=httpx.Timeout(0.8),limits=httpx.Limits(max_connections=6,max_keepalive_connections=6),transport=transport,trust_env=False,follow_redirects=False,)asclient:app.state.service=QuoteService(client)yieldapp=FastAPI(title="QuoteFlow",version="1.0.0",lifespan=lifespan)@app.get("/health")asyncdefhealth():return{"status":"ok"}@app.post("/api/quotes",response_model=QuoteResult)asyncdefquotes(query:QuoteRequest,request:Request):try:returnawaitrequest.app.state.service.quote(query)exceptBusyasexc:raiseHTTPException(503,"capacity_exhausted",headers={"Retry-After":"1"})fromexcreturnapp app=create_app()
