Python进阶与并发编程14：boreal偶尔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_data；404同样只一次。这组测试能发现把尝试数写成三次、每次换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.exe，Linux用.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 /reset；GET /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.py

"""E13 答案：一次拒绝之后容量仍能归还、复用。"""importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/"checkpoints/13/src"))fromquoteflow.modelsimportQuoteRequestfromquoteflow.serviceimportBusy,QuoteServicefromquoteflow.apiimportcreate_appasyncdefmain():entered,release=asyncio.Event(),asyncio.Event()asyncdefsupplier(request):entered.set()awaitrelease.wait()name=request.url.path.rsplit("/",1)[1]returnhttpx.Response(200,json=dict(supplier=name,sku="KB-001",unit_cents=100,shipping_cents=0,stock=100))asyncwithhttpx.AsyncClient(transport=httpx.MockTransport(supplier),base_url="http://provider")asclient:service=QuoteService(client,inbound_limit=1)first=asyncio.create_task(service.quote(QuoteRequest()))awaitentered.wait()try:awaitservice.quote(QuoteRequest())exceptBusy:passelse:raiseAssertionError("超过入站容量必须拒绝")release.set()assert(awaitfirst).bestisnotNoneassertservice.active_requests==0assert(awaitservice.quote(QuoteRequest())).bestisnotNoneassertservice.active_requests==0# HTTP层单独验证：Busy映射503，供应商503则是200响应中的失败项。asyncdefunavailable(request):returnhttpx.Response(503)app=create_app(transport=httpx.MockTransport(unavailable))asyncwithapp.router.lifespan_context(app):shared=app.state.service shared.inbound_limit=1asyncwithhttpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url="http://api")asbrowser:withshared.admission():rejected=awaitbrowser.post("/api/quotes",json={"quantity":2})assertrejected.status_code==503assertrejected.headers["Retry-After"]=="1"recovered=awaitbrowser.post("/api/quotes",json={"quantity":2})assertrecovered.status_code==200data=recovered.json()assertdata["best"]isNoneassertlen(data["failures"])==3assert{failure["code"]forfailureindata["failures"]}=={"http_503"}assertshared.active_requests==0print("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 = ["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/14/src/quoteflow/__init__.py

"""Standalone course checkpoint."""

checkpoints/14/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:str

checkpoints/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: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):forattemptinrange(2):retryable=Falsetry:# 每次 HTTP 尝试才占出站槽位，退避睡眠必须先释放槽位。asyncwithself.outbound:response=awaitself.client.get(f"/quotes/{supplier}",params={"sku":query.sku,"scenario":query.scenario},headers={"X-Request-ID":request_id},)ifresponse.status_code!=200:failure=Failure(supplier=supplier,code=f"http_{response.status_code}")retryable=response.status_code==503else: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")returnquoteexcepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生；这是坏数据，不能重试。returnFailure(supplier=supplier,code="invalid_data")exceptTRANSIENT:failure=Failure(supplier=supplier,code="transport_error")retryable=Trueifnotretryableorattempt==1:returnfailureawaitasyncio.sleep(0.02+random.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: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/14/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()