Python进阶与并发编程14boreal偶尔503重试一次究竟要花掉谁的预算假设boreal报价最低但某次调用正好遇到服务切换第一次返回503紧接着就恢复。第13课会保留其他供应商的结果因此查询没有完全失败却错失了更便宜的报价。业务希望“再试一下”。这听起来只需要把HTTP请求放进一个循环真正危险的是循环会增加出站流量、延长响应时间并在故障最严重的时候给供应商追加压力。本课把重试限制为每供应商至多两次尝试只有503与明确列举的瞬态传输异常能触发第二次。共享出站6、入站8和0.8秒总预算全部保持。代码在checkpoints/14仍只有单项接口。请先理解这张预算账单再考虑把重试放进你自己的服务。E13答案被拒绝的请求不能偷走一个槽位完整答案lessons/14/answer_previous.py从第13课阶段导入QuoteService运行python lessons/14/answer_previous.py。第一条请求的供应商处理器设置entered事件然后等待release。测试等entered之后再发第二条确保此时唯一入站槽位确实被占用第二条应抛Busy而不是悄悄排队。放行release之后等待第一条的结果再检查active_requests为0最后发送第三条并成功。这个顺序同时发现三种问题忙碌时没有拒绝拒绝时错误递增计数请求完成时没有在finally中归还容量。验证槽位复用是对用户可感知行为的断言不是机械检查代码写了几次加减。在HTTP层入口把Busy变为503并设置Retry-After:1。这个头表达客户端可以稍后重试不构成“等待一秒就一定能成功”的保证。自动重试行为由客户端自己决定客户端如果无视容量信号、所有请求立即齐刷刷重发服务仍然会过载。这也是本课服务端退避需要少量抖动的原因。答案的第二段实际进入第13课FastAPI的lifespan通过ASGITransport提交请求。先占用唯一入站槽位断言响应为503且Retry-After为1释放槽位后让供应商本身返回503断言接口此时是200、best为空、三项failures为http_503。这使“本服务拒绝准入”和“供应商查询失败”两个不同位置的503得到完整验证。先问操作是否适合重试再问异常类型QuoteFlow执行的是读取报价的GET本次第二次尝试不应创建订单或重复扣款。这样的读取操作允许我们讨论有限重试。如果把同样代码搬到支付POST即使客户端没有收到响应服务器也可能已经完成扣款传输错误只说明我们不知道结果不能证明远端没有执行。那种业务需要幂等键、结果查询或其他明确协议本课的读取重试不能当作通用事务恢复方案。即便是GET也应考虑供应商成本、限额和服务条款。两次尝试意味着故障持续时每个逻辑调用的流量可能翻倍三家供应商的查询原本最多三次HTTP本课在都可重试失败时最多六次HTTP。固定两次让放大倍数有上限而总预算进一步限制慢故障时能够真正发出的尝试数量。不能说“最多两次所以肯定会试满两次”第一轮已用完预算第二轮就没有资格开始。请求ID在一次供应商查询的所有尝试中保持一致。实验服务器的flaky规则是同一个X-Request-ID第一次503、第二次正常如果每次循环新建ID服务器会把每次都视为第一次结果永远失败。这个ID帮助关联同一逻辑查询的尝试不等于已经实现支付级幂等也不能仅凭一个随机字符串就宣称全链路追踪已经完整。错误分类的目的是决定下一步而不是收集名称HTTP503属于当前供应商暂时不可用本课程允许重试。404表示这个供应商没有对应资源重复同一读取通常不能凭空产生SKU因此立即返回http_404。其他HTTP状态本实现也直接报告不假设每一种5xx或429都适合无条件重试。若将来支持Retry-After、429或某些500应分别为供应商合同与等待上限写测试而非简单扩大一个状态码范围。200之后的畸形JSON、字段类型错误、负金额、响应身份或SKU错配都属于invalid_data重试不会自动修复协议。这些错误必须保留可见证据让供应商或我们修正合同。库存不足更不是重试信号它是一份已经读到的有效业务结果过滤即可。否则一个库存为1的cedar会在购买2件时无意义地消耗两次网络操作。坏gzip是发生得更早的一种数据错误HTTPX读取响应正文时就会抛DecodingError还没有机会执行response.json。它不是TransportError的子类我们在client.get外单独捕获并立即返回invalid_data绝不加入TRANSIENT集合。测试返回真实的错误压缩字节让HTTPX自己解码断言boreal只访问一次、另外两家仍保留、HTTP入口仍为200并验证响应流和容量归还。传输异常也没有一锅端地重试。TRANSIENT显式列出ConnectError/ConnectTimeout、ReadError/ReadTimeout、WriteError/WriteTimeout、RemoteProtocolError、PoolTimeout。它们表达建立连接或读取过程出现暂时障碍在这项只读操作上允许第二次尝试。LocalProtocolError等本地协议构造问题未列入常常需要改代码而非增加流量。不能因为它们都继承某个HTTP异常父类就赋予相同恢复策略。对于测试刻意抛出的RuntimeError服务不会返回一个看似正常的transport_error。它由TaskGroup按程序异常传播入站finally仍清理资源。故障分类与资源清理是两件事我们可以拒绝掩盖代码错误同时仍保证它不会泄漏容量。一次尝试用完槽位退避必须站到队伍外面for attempt in range(2)表明最多两次。每一轮HTTP访问用async with self.outbound包住离开这个上下文后才判断是否可重试、是否还有次数最后await短暂退避。将sleep放进Semaphore内部会让一个暂时什么也没做的失败请求霸占槽位阻止健康供应商执行。下游故障时这种浪费尤其明显。这里退避采用20–30毫秒随机抖动目的是让同时失败的请求不要完全同步重发。只有一次重试因此没有必要建立复杂的指数退避配置框架若有更多尝试指数增长与上限会更重要但尝试数增加也会改变整体合同。抖动不是保证公平的调度算法它只是减少整齐重发的机会。本课保留0.8秒的绝对deadline。第一次调用、出站排队、释放槽位后的sleep、第二次再排队与HTTP执行全在_collect外层同一个timeout_at里。退避没有另开计时器给自己延长生命。若第一次已用0.79秒20毫秒sleep可能直接被总预算取消结果是timeout而不是“为了凑满两次再拖延一轮”。HTTP层自己的timeout仍存在用于限制网络阶段它可能先抛出ReadTimeout服务按瞬态故障尝试第二轮。但剩余总预算很短时外层取消会先到最终报告timeout。两种超时的差别在于谁规定边界一个是客户端网络阶段设置一个是整个业务请求的剩余预算。读日志时必须同时理解它们而不是把所有等待异常混成一个原因。为什么失败的结果仍能保持稳定两次503之后代码返回最后一轮的http_503第一次ConnectError、第二次正常则返回正常报价第一次503、第二次invalid_data则立即返回invalid_data不能继续开第三次。每家供应商独立执行这个决策atlas正常无需等待boreal决定是否重试才开始自己的请求。所有供应商完成后依然按总价、supplier排序结果不会因为第二次请求恰好更快完成而改变展示规则。completed只有在供应商得到最终值时才写入尚在重试或退避的供应商到总截止时刻会成为timeout。已经成功的报价保留。这个关系让调用方能分别判断“目前有最低可用报价”与“仍有供应商因预算没有完成”避免把部分成功误读成已经穷尽所有市场报价。随机抖动也提醒我们不要给测试写死精确耗时。业务测试应断言次数、分类、可售结果与宽容的总时间上限性能测量则重复运行、报告分布。本课不是把20毫秒当成任何真实供应商都最优的参数而是给出可观察、可修改、受预算约束的最小策略。分别做协议测试和真实HTTP实验传输重试另设四条回归路径ConnectError与ReadTimeout各自测试“首次失败、次轮恢复”和“连续失败两轮”。它们都断言boreal恰好两次且X-Request-ID不变恢复时boreal重新进入quotes持续失败时报告transport_error同时atlas和cedar保留。这样不会把503重试通过误当成所有瞬态异常分支也已得到覆盖。阶段测试让MockTransport按照supplier与request_id计数。flaky第一次503、第二次返回200断言boreal恰好两次且两次ID相同broken始终503断言最多两次malformed字段为字符串断言一次即invalid_data404同样只一次。这组测试能发现把尝试数写成三次、每次换ID、或把数据校验失败也重试的回归。它们并未经过网络因此还需要真实实验。供应商启动命令是python tools/provider_lab.py --port 9101。在独立checkpoints/14环境中执行python -m pip install -e .[test]、python -m pytest -q再启动python -m uvicorn quoteflow.api:app --host 127.0.0.1 --port 9100。Windows用Python3.11创建的.venv\Scripts\python.exeLinux用.venv/bin/python。Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:flaky}Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:malformed}第一条通常恢复boreal最低报价第二条保留atlas并报告boreal的invalid_data。每轮统计前要等供应商active归零再调用POST /resetGET /stats的per_supplier能提供真实请求数。这个控制接口只适合本机教学服务器不能把可重置全局统计的接口原样公开到真实供应商环境。评审一个重试循环可以先做四次纸上演算第一种路径是立即200总共一次HTTP没有sleep没有第二次领取槽位。第二种路径是503后200先释放首次槽位退避后重新竞争第二轮成功即返回。第三种路径是503后503两轮结束返回http_503不存在第三轮。第四种路径是503后在退避中到期外层timeout取消sleep收集器把未完成的boreal记为timeout。四条路径的资源释放位置相同退出原因却各不相同。如果读者把sleep挪到HTTP槽位内部这四次纸上演算会立刻显示第二到第四条都多占了一段无效容量。如果把deadline放到循环里面就会显示第三条可能获得两份完整时间预算。如果捕获CancelledError后continue第四条甚至可能越过总截止时刻继续尝试。这样评审代码是在追踪一条请求的实际状态变化而不是凭“用了async with所以一定安全”的印象。重试策略还需要和可观察性配套。响应保留request_id测试统计每个ID的尝试次数真实部署若要解释一次故障应记录供应商、尝试序号、分类与剩余预算但不应随意记录用户敏感数据。课程暂未引入一整套日志平台并不意味着观测不重要而是先把可解释的请求身份和失败代码留在合同里让后续工具有可靠基础。先理解失败再决定重试才能让额外的请求真正服务于恢复。E14恢复一次与坚持错误只差一条规则编写一个可控的flaky供应商按同一个X-Request-ID首轮503、次轮正常断言boreal进入quotes且计数为2随后提交malformed断言boreal为invalid_data、计数为1、其他健康报价仍存在。不要只断言best非空因为即使完全没有实现重试atlas也可能让这个断言通过。完整答案将在第15课的lessons/15/answer_previous.py。尝试再说明为何没有给所有HTTP错误重试为何退避必须在信号量外为何两次尝试仍共用原deadline。这三句话能讲清才说明你掌握了机制的适用边界而不仅是会把请求放进for循环。核心模块全文附后。模型与入口保持上课合同service新增尝试循环和明确的瞬态异常集合项目依旧是本地课程服务尚不具备鉴权、跨进程限流、完备日志与线上治理体系。本课完整代码lessons/14/answer_previous.pyE13 答案一次拒绝之后容量仍能归还、复用。importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/checkpoints/13/src))fromquoteflow.modelsimportQuoteRequestfromquoteflow.serviceimportBusy,QuoteServicefromquoteflow.apiimportcreate_appasyncdefmain():entered,releaseasyncio.Event(),asyncio.Event()asyncdefsupplier(request):entered.set()awaitrelease.wait()namerequest.url.path.rsplit(/,1)[1]returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock100))asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(supplier),base_urlhttp://provider)asclient:serviceQuoteService(client,inbound_limit1)firstasyncio.create_task(service.quote(QuoteRequest()))awaitentered.wait()try:awaitservice.quote(QuoteRequest())exceptBusy:passelse:raiseAssertionError(超过入站容量必须拒绝)release.set()assert(awaitfirst).bestisnotNoneassertservice.active_requests0assert(awaitservice.quote(QuoteRequest())).bestisnotNoneassertservice.active_requests0# HTTP层单独验证Busy映射503供应商503则是200响应中的失败项。asyncdefunavailable(request):returnhttpx.Response(503)appcreate_app(transporthttpx.MockTransport(unavailable))asyncwithapp.router.lifespan_context(app):sharedapp.state.service shared.inbound_limit1asyncwithhttpx.AsyncClient(transporthttpx.ASGITransport(appapp),base_urlhttp://api)asbrowser:withshared.admission():rejectedawaitbrowser.post(/api/quotes,json{quantity:2})assertrejected.status_code503assertrejected.headers[Retry-After]1recoveredawaitbrowser.post(/api/quotes,json{quantity:2})assertrecovered.status_code200datarecovered.json()assertdata[best]isNoneassertlen(data[failures])3assert{failure[code]forfailureindata[failures]}{http_503}assertshared.active_requests0print(E13 passed: rejection does not consume capacity; slot reused; HTTP 503/Retry-After verified)if__name____main__:asyncio.run(main())checkpoints/14/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/14/src/quoteflow/__init__.pyStandalone course checkpoint.checkpoints/14/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/14/src/quoteflow/service.py每进程一个服务实例所有 HTTP handler 共用准入计数与出站信号量。importasyncioimportrandomfromcontextlibimportcontextmanagerfromtimeimportperf_counterfromuuidimportuuid4importhttpxfrompydanticimportValidationErrorfrom.modelsimportFailure,QuoteRequest,QuoteResult,RankedQuote,SupplierQuote SUPPLIERS(atlas,boreal,cedar)TRANSIENT(httpx.ConnectError,httpx.ConnectTimeout,httpx.ReadError,httpx.ReadTimeout,httpx.WriteError,httpx.WriteTimeout,httpx.RemoteProtocolError,httpx.PoolTimeout,)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):forattemptinrange(2):retryableFalsetry:# 每次 HTTP 尝试才占出站槽位退避睡眠必须先释放槽位。asyncwithself.outbound:responseawaitself.client.get(f/quotes/{supplier},params{sku:query.sku,scenario:query.scenario},headers{X-Request-ID:request_id},)ifresponse.status_code!200:failureFailure(suppliersupplier,codefhttp_{response.status_code})retryableresponse.status_code503else:try:quoteSupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(suppliersupplier,codeinvalid_data)ifquote.supplier!supplierorquote.sku!query.sku:returnFailure(suppliersupplier,codeinvalid_data)returnquoteexcepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生这是坏数据不能重试。returnFailure(suppliersupplier,codeinvalid_data)exceptTRANSIENT:failureFailure(suppliersupplier,codetransport_error)retryableTrueifnotretryableorattempt1:returnfailureawaitasyncio.sleep(0.02random.uniform(0,0.01))raiseAssertionError(unreachable)asyncdef_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/14/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() Python进阶与并发编程14：boreal偶尔503，重试一次究竟要花掉谁的预算
Python进阶与并发编程14boreal偶尔503重试一次究竟要花掉谁的预算假设boreal报价最低但某次调用正好遇到服务切换第一次返回503紧接着就恢复。第13课会保留其他供应商的结果因此查询没有完全失败却错失了更便宜的报价。业务希望“再试一下”。这听起来只需要把HTTP请求放进一个循环真正危险的是循环会增加出站流量、延长响应时间并在故障最严重的时候给供应商追加压力。本课把重试限制为每供应商至多两次尝试只有503与明确列举的瞬态传输异常能触发第二次。共享出站6、入站8和0.8秒总预算全部保持。代码在checkpoints/14仍只有单项接口。请先理解这张预算账单再考虑把重试放进你自己的服务。E13答案被拒绝的请求不能偷走一个槽位完整答案lessons/14/answer_previous.py从第13课阶段导入QuoteService运行python lessons/14/answer_previous.py。第一条请求的供应商处理器设置entered事件然后等待release。测试等entered之后再发第二条确保此时唯一入站槽位确实被占用第二条应抛Busy而不是悄悄排队。放行release之后等待第一条的结果再检查active_requests为0最后发送第三条并成功。这个顺序同时发现三种问题忙碌时没有拒绝拒绝时错误递增计数请求完成时没有在finally中归还容量。验证槽位复用是对用户可感知行为的断言不是机械检查代码写了几次加减。在HTTP层入口把Busy变为503并设置Retry-After:1。这个头表达客户端可以稍后重试不构成“等待一秒就一定能成功”的保证。自动重试行为由客户端自己决定客户端如果无视容量信号、所有请求立即齐刷刷重发服务仍然会过载。这也是本课服务端退避需要少量抖动的原因。答案的第二段实际进入第13课FastAPI的lifespan通过ASGITransport提交请求。先占用唯一入站槽位断言响应为503且Retry-After为1释放槽位后让供应商本身返回503断言接口此时是200、best为空、三项failures为http_503。这使“本服务拒绝准入”和“供应商查询失败”两个不同位置的503得到完整验证。先问操作是否适合重试再问异常类型QuoteFlow执行的是读取报价的GET本次第二次尝试不应创建订单或重复扣款。这样的读取操作允许我们讨论有限重试。如果把同样代码搬到支付POST即使客户端没有收到响应服务器也可能已经完成扣款传输错误只说明我们不知道结果不能证明远端没有执行。那种业务需要幂等键、结果查询或其他明确协议本课的读取重试不能当作通用事务恢复方案。即便是GET也应考虑供应商成本、限额和服务条款。两次尝试意味着故障持续时每个逻辑调用的流量可能翻倍三家供应商的查询原本最多三次HTTP本课在都可重试失败时最多六次HTTP。固定两次让放大倍数有上限而总预算进一步限制慢故障时能够真正发出的尝试数量。不能说“最多两次所以肯定会试满两次”第一轮已用完预算第二轮就没有资格开始。请求ID在一次供应商查询的所有尝试中保持一致。实验服务器的flaky规则是同一个X-Request-ID第一次503、第二次正常如果每次循环新建ID服务器会把每次都视为第一次结果永远失败。这个ID帮助关联同一逻辑查询的尝试不等于已经实现支付级幂等也不能仅凭一个随机字符串就宣称全链路追踪已经完整。错误分类的目的是决定下一步而不是收集名称HTTP503属于当前供应商暂时不可用本课程允许重试。404表示这个供应商没有对应资源重复同一读取通常不能凭空产生SKU因此立即返回http_404。其他HTTP状态本实现也直接报告不假设每一种5xx或429都适合无条件重试。若将来支持Retry-After、429或某些500应分别为供应商合同与等待上限写测试而非简单扩大一个状态码范围。200之后的畸形JSON、字段类型错误、负金额、响应身份或SKU错配都属于invalid_data重试不会自动修复协议。这些错误必须保留可见证据让供应商或我们修正合同。库存不足更不是重试信号它是一份已经读到的有效业务结果过滤即可。否则一个库存为1的cedar会在购买2件时无意义地消耗两次网络操作。坏gzip是发生得更早的一种数据错误HTTPX读取响应正文时就会抛DecodingError还没有机会执行response.json。它不是TransportError的子类我们在client.get外单独捕获并立即返回invalid_data绝不加入TRANSIENT集合。测试返回真实的错误压缩字节让HTTPX自己解码断言boreal只访问一次、另外两家仍保留、HTTP入口仍为200并验证响应流和容量归还。传输异常也没有一锅端地重试。TRANSIENT显式列出ConnectError/ConnectTimeout、ReadError/ReadTimeout、WriteError/WriteTimeout、RemoteProtocolError、PoolTimeout。它们表达建立连接或读取过程出现暂时障碍在这项只读操作上允许第二次尝试。LocalProtocolError等本地协议构造问题未列入常常需要改代码而非增加流量。不能因为它们都继承某个HTTP异常父类就赋予相同恢复策略。对于测试刻意抛出的RuntimeError服务不会返回一个看似正常的transport_error。它由TaskGroup按程序异常传播入站finally仍清理资源。故障分类与资源清理是两件事我们可以拒绝掩盖代码错误同时仍保证它不会泄漏容量。一次尝试用完槽位退避必须站到队伍外面for attempt in range(2)表明最多两次。每一轮HTTP访问用async with self.outbound包住离开这个上下文后才判断是否可重试、是否还有次数最后await短暂退避。将sleep放进Semaphore内部会让一个暂时什么也没做的失败请求霸占槽位阻止健康供应商执行。下游故障时这种浪费尤其明显。这里退避采用20–30毫秒随机抖动目的是让同时失败的请求不要完全同步重发。只有一次重试因此没有必要建立复杂的指数退避配置框架若有更多尝试指数增长与上限会更重要但尝试数增加也会改变整体合同。抖动不是保证公平的调度算法它只是减少整齐重发的机会。本课保留0.8秒的绝对deadline。第一次调用、出站排队、释放槽位后的sleep、第二次再排队与HTTP执行全在_collect外层同一个timeout_at里。退避没有另开计时器给自己延长生命。若第一次已用0.79秒20毫秒sleep可能直接被总预算取消结果是timeout而不是“为了凑满两次再拖延一轮”。HTTP层自己的timeout仍存在用于限制网络阶段它可能先抛出ReadTimeout服务按瞬态故障尝试第二轮。但剩余总预算很短时外层取消会先到最终报告timeout。两种超时的差别在于谁规定边界一个是客户端网络阶段设置一个是整个业务请求的剩余预算。读日志时必须同时理解它们而不是把所有等待异常混成一个原因。为什么失败的结果仍能保持稳定两次503之后代码返回最后一轮的http_503第一次ConnectError、第二次正常则返回正常报价第一次503、第二次invalid_data则立即返回invalid_data不能继续开第三次。每家供应商独立执行这个决策atlas正常无需等待boreal决定是否重试才开始自己的请求。所有供应商完成后依然按总价、supplier排序结果不会因为第二次请求恰好更快完成而改变展示规则。completed只有在供应商得到最终值时才写入尚在重试或退避的供应商到总截止时刻会成为timeout。已经成功的报价保留。这个关系让调用方能分别判断“目前有最低可用报价”与“仍有供应商因预算没有完成”避免把部分成功误读成已经穷尽所有市场报价。随机抖动也提醒我们不要给测试写死精确耗时。业务测试应断言次数、分类、可售结果与宽容的总时间上限性能测量则重复运行、报告分布。本课不是把20毫秒当成任何真实供应商都最优的参数而是给出可观察、可修改、受预算约束的最小策略。分别做协议测试和真实HTTP实验传输重试另设四条回归路径ConnectError与ReadTimeout各自测试“首次失败、次轮恢复”和“连续失败两轮”。它们都断言boreal恰好两次且X-Request-ID不变恢复时boreal重新进入quotes持续失败时报告transport_error同时atlas和cedar保留。这样不会把503重试通过误当成所有瞬态异常分支也已得到覆盖。阶段测试让MockTransport按照supplier与request_id计数。flaky第一次503、第二次返回200断言boreal恰好两次且两次ID相同broken始终503断言最多两次malformed字段为字符串断言一次即invalid_data404同样只一次。这组测试能发现把尝试数写成三次、每次换ID、或把数据校验失败也重试的回归。它们并未经过网络因此还需要真实实验。供应商启动命令是python tools/provider_lab.py --port 9101。在独立checkpoints/14环境中执行python -m pip install -e .[test]、python -m pytest -q再启动python -m uvicorn quoteflow.api:app --host 127.0.0.1 --port 9100。Windows用Python3.11创建的.venv\Scripts\python.exeLinux用.venv/bin/python。Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:flaky}Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:malformed}第一条通常恢复boreal最低报价第二条保留atlas并报告boreal的invalid_data。每轮统计前要等供应商active归零再调用POST /resetGET /stats的per_supplier能提供真实请求数。这个控制接口只适合本机教学服务器不能把可重置全局统计的接口原样公开到真实供应商环境。评审一个重试循环可以先做四次纸上演算第一种路径是立即200总共一次HTTP没有sleep没有第二次领取槽位。第二种路径是503后200先释放首次槽位退避后重新竞争第二轮成功即返回。第三种路径是503后503两轮结束返回http_503不存在第三轮。第四种路径是503后在退避中到期外层timeout取消sleep收集器把未完成的boreal记为timeout。四条路径的资源释放位置相同退出原因却各不相同。如果读者把sleep挪到HTTP槽位内部这四次纸上演算会立刻显示第二到第四条都多占了一段无效容量。如果把deadline放到循环里面就会显示第三条可能获得两份完整时间预算。如果捕获CancelledError后continue第四条甚至可能越过总截止时刻继续尝试。这样评审代码是在追踪一条请求的实际状态变化而不是凭“用了async with所以一定安全”的印象。重试策略还需要和可观察性配套。响应保留request_id测试统计每个ID的尝试次数真实部署若要解释一次故障应记录供应商、尝试序号、分类与剩余预算但不应随意记录用户敏感数据。课程暂未引入一整套日志平台并不意味着观测不重要而是先把可解释的请求身份和失败代码留在合同里让后续工具有可靠基础。先理解失败再决定重试才能让额外的请求真正服务于恢复。E14恢复一次与坚持错误只差一条规则编写一个可控的flaky供应商按同一个X-Request-ID首轮503、次轮正常断言boreal进入quotes且计数为2随后提交malformed断言boreal为invalid_data、计数为1、其他健康报价仍存在。不要只断言best非空因为即使完全没有实现重试atlas也可能让这个断言通过。完整答案将在第15课的lessons/15/answer_previous.py。尝试再说明为何没有给所有HTTP错误重试为何退避必须在信号量外为何两次尝试仍共用原deadline。这三句话能讲清才说明你掌握了机制的适用边界而不仅是会把请求放进for循环。核心模块全文附后。模型与入口保持上课合同service新增尝试循环和明确的瞬态异常集合项目依旧是本地课程服务尚不具备鉴权、跨进程限流、完备日志与线上治理体系。本课完整代码lessons/14/answer_previous.pyE13 答案一次拒绝之后容量仍能归还、复用。importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/checkpoints/13/src))fromquoteflow.modelsimportQuoteRequestfromquoteflow.serviceimportBusy,QuoteServicefromquoteflow.apiimportcreate_appasyncdefmain():entered,releaseasyncio.Event(),asyncio.Event()asyncdefsupplier(request):entered.set()awaitrelease.wait()namerequest.url.path.rsplit(/,1)[1]returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock100))asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(supplier),base_urlhttp://provider)asclient:serviceQuoteService(client,inbound_limit1)firstasyncio.create_task(service.quote(QuoteRequest()))awaitentered.wait()try:awaitservice.quote(QuoteRequest())exceptBusy:passelse:raiseAssertionError(超过入站容量必须拒绝)release.set()assert(awaitfirst).bestisnotNoneassertservice.active_requests0assert(awaitservice.quote(QuoteRequest())).bestisnotNoneassertservice.active_requests0# HTTP层单独验证Busy映射503供应商503则是200响应中的失败项。asyncdefunavailable(request):returnhttpx.Response(503)appcreate_app(transporthttpx.MockTransport(unavailable))asyncwithapp.router.lifespan_context(app):sharedapp.state.service shared.inbound_limit1asyncwithhttpx.AsyncClient(transporthttpx.ASGITransport(appapp),base_urlhttp://api)asbrowser:withshared.admission():rejectedawaitbrowser.post(/api/quotes,json{quantity:2})assertrejected.status_code503assertrejected.headers[Retry-After]1recoveredawaitbrowser.post(/api/quotes,json{quantity:2})assertrecovered.status_code200datarecovered.json()assertdata[best]isNoneassertlen(data[failures])3assert{failure[code]forfailureindata[failures]}{http_503}assertshared.active_requests0print(E13 passed: rejection does not consume capacity; slot reused; HTTP 503/Retry-After verified)if__name____main__:asyncio.run(main())checkpoints/14/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/14/src/quoteflow/__init__.pyStandalone course checkpoint.checkpoints/14/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/14/src/quoteflow/service.py每进程一个服务实例所有 HTTP handler 共用准入计数与出站信号量。importasyncioimportrandomfromcontextlibimportcontextmanagerfromtimeimportperf_counterfromuuidimportuuid4importhttpxfrompydanticimportValidationErrorfrom.modelsimportFailure,QuoteRequest,QuoteResult,RankedQuote,SupplierQuote SUPPLIERS(atlas,boreal,cedar)TRANSIENT(httpx.ConnectError,httpx.ConnectTimeout,httpx.ReadError,httpx.ReadTimeout,httpx.WriteError,httpx.WriteTimeout,httpx.RemoteProtocolError,httpx.PoolTimeout,)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):forattemptinrange(2):retryableFalsetry:# 每次 HTTP 尝试才占出站槽位退避睡眠必须先释放槽位。asyncwithself.outbound:responseawaitself.client.get(f/quotes/{supplier},params{sku:query.sku,scenario:query.scenario},headers{X-Request-ID:request_id},)ifresponse.status_code!200:failureFailure(suppliersupplier,codefhttp_{response.status_code})retryableresponse.status_code503else:try:quoteSupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(suppliersupplier,codeinvalid_data)ifquote.supplier!supplierorquote.sku!query.sku:returnFailure(suppliersupplier,codeinvalid_data)returnquoteexcepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生这是坏数据不能重试。returnFailure(suppliersupplier,codeinvalid_data)exceptTRANSIENT:failureFailure(suppliersupplier,codetransport_error)retryableTrueifnotretryableorattempt1:returnfailureawaitasyncio.sleep(0.02random.uniform(0,0.01))raiseAssertionError(unreachable)asyncdef_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/14/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()