Python进阶与并发编程11请求取消之后资源真的归还了吗第09课已经能在预算到期时取消慢供应商第10课能用队列限制生产速度。现在换一个触发来源调用方不再需要结果主动取消整个批次。屏幕上出现一个CancelledError并不等于系统已经恢复健康如果信号量槽位没归还、工作者还在等待、请求ID污染了下一次调用下一批工作仍会出问题。本课用一个小型资源门控实验把取消过程拆开观察。共享资源只有两个槽位六项工作争抢当前批次取消后下一批继续复用同一个门控对象。我们用“下一批还能完成”证明槽位可用不去读取Semaphore的私有内部字段。这比只写一句finally更加接近服务需要的保证。E10答案队列顺序来自索引不来自等待时间上一课E10要求容量2、两个固定消费者处理[5,4,3,2,1,0]每项等待与数值成比例的时间再返回平方。answer_previous.py把这些输入交给随课pipeline.run_queue断言结果为[25,16,9,4,1,0]最大队列长度不超过2完成业务项数6。慢的5不一定比快的4先完成但两者都把答案写回原索引所以结果可靠。队列的哨兵不计入业务completed却仍需task_doneTaskGroup等待生产者和消费者结束后run_queue才返回。第10课的队列核心原样复制到本课使答案独立运行。cancel只提出请求await才观察到结束task.cancel()请求向目标协程注入CancelledError。它不会像强制杀进程那样立即移除执行栈。任务通常在下一个可中断的await处感受到取消然后执行异常展开和finally清理。调用cancel之后应继续await目标任务并在需要的位置识别取消传播。假设先task.cancel()立刻断言活动计数为0测试可能偶尔失败。那不是一定漏清理而是目标任务还没有被调度去处理取消。正确顺序是请求取消、等待目标结束、再检查资源状态。若不等待任务里的异常可能无人领取应用也无法知道何时可以关闭共享客户端。取消与TaskGroup配合时父任务取消会向其组内子任务传播任务组等待子任务退出后父任务的取消继续向上。这样调用者await批次任务结束意味着该组内工作也已经收尾。任务所有权形成一条链避免“上层返回了下层还在使用已关闭资源”。槽位的获取和释放必须成对出现ResourceGate内部创建asyncio.Semaphore(limit)。slot是异步上下文管理器先进入async with self.semaphore取得槽位再增加active执行工作最后在finally减少active信号量上下文负责释放槽位。这个结构区分两类取消位置。等待信号量时被取消任务根本没取得槽位不应减active或手工release已经取得槽位、正在工作时被取消则必须减少active并释放。若把release放在一个无条件finally里但没有记录是否获取成功就可能“没拿到也归还”把容量越用越大。异步上下文管理器让这对责任绑定在一起。active与peak的更新中没有await且本实验只在同一事件循环线程中使用所以计数不需要线程锁。它们是便于教学与观察的指标真正限制并发的仍是Semaphore不能反过来通过轮询active自己实现锁。这里的item * 2只是生命周期实验的可见输出不是最终完整报价项目。真实报价中的对应关系是槽位内执行HTTP调用结果写回报价行取消时释放出站容量。本实验一次为每个输入创建Task适合有限小批次。信号量只限制活跃工作不限制排队Task数量大规模输入应采用第10课的有界队列与固定工作者。工作用await asyncio.sleep(delay)模拟可取消的等待目的在于把外部HTTP服务的行为从本次实验中移开。前几课已经验证真实HTTP这里要证明取消发生在不同槽位位置时门控对象能否恢复。下一阶段会把同样的结构放到共享出站HTTP请求周围。不要把取消伪装成一个成功返回在run_batch里业务结果只在所有工作完成后返回。finally恢复上下文无论成功、异常还是取消都执行但不return。若在finally里return results会覆盖本来应该传播的异常调用方可能得到一个半填充列表并以为批次成功。Python3.11中CancelledError继承BaseException。我们不捕获它来构造普通失败行也不在任务内部写“取消就返回None”。如果为了日志必须捕获应在完成必要记录后使用裸raise重新抛出。否则TaskGroup、timeout和上层关闭流程对取消的假设都会被破坏。asyncio.shield也不能作为默认修补方法。它可能让外层等待者取消而被保护任务继续运行如果没人保存并等待那个任务反而重新引入后台遗留工作。只有那些确实需要完成且有明确所有者的短收尾动作才值得讨论屏蔽取消和额外期限。本课资源释放只需同步计数与上下文退出不需要shield。清理代码也有时间和失败边界取消不是硬实时中断。事件循环正在执行长CPU循环时没有机会交出控制权cancel只能等待底层阻塞线程已经开始运行时取消协程也无法杀掉它。若finally需要等待网络关闭或刷新数据实际退出时间还会包含清理耗时。因此“取消后资源归还”应定义为await取消任务完成之后的条件而不是cancel调用一返回就满足。生产服务中的异步清理也应有合理边界不能为了释放资源又无限等待一个坏网络。客户端库通常负责自己的连接关闭协议应用负责何时调用和等待关闭不能随意把资源生命周期拆散。还要区分本地取消与远端停止。关闭HTTP等待通常释放本地连接占用但远端服务器可能已经接到请求并继续计算。我们的教学供应商收到慢请求后可能还在自己的线程中睡眠本地槽位归还不等于它的active统计瞬间归零。只读报价便于安全重试修改库存或付款等操作则需要幂等键和业务协议不能把取消当作撤销交易。为什么请求ID不用普通全局变量两个并发批次A和B都需要在日志里标记自己的request_id。如果把它写入普通模块级变量A设置之后遇到awaitB可能把变量改成BA恢复运行时就会读到错误ID。线程本地变量也不适合多个协程共享同一个事件循环线程的情形。ContextVar把值放在当前执行上下文中。run_batch调用token request_id.set(batch_id)随后创建子任务。任务在创建时获取当前上下文的副本因此每个子任务都能读到这个批次的ID并发批次彼此独立。子任务之后自行set值也不会自动把新值写回父任务或其他兄弟任务。这里的“副本”不是深复制所有对象。如果ContextVar里放的是一个可变字典多个上下文可能仍持有同一个字典引用对字典内容的修改可能泄漏。请求ID选择不可变字符串避免引入共享可变状态。ContextVar解决上下文定位不是万能的隔离容器。token.reset恢复的是之前的状态set返回的token记住当前上下文中先前的值finally里request_id.reset(token)将它恢复。不要无条件request_id.set(-)因为调用方可能已经设置了一个外层追踪ID。嵌套调用结束后应回到调用前状态而不是回到你猜测的默认值。token应在对应的上下文使用且只reset一次。把token传给另一个任务去恢复或者把“清理token”的责任交给子任务都会混淆上下文归属。谁set谁在同一调用范围的finally里reset最容易检查正确性。本课main默认调用正常批次输出三项value为2、4、6request_id都为demo结束后active_after为0peak为2调用方request_id恢复为-。另一项测试同时运行A和B批次分别检查各自行的ID并检查外层默认值未改变。这样验证的是任务间上下文隔离而不是只测试一次set/get。让测试证明下一次操作仍然可用取消测试创建共享ResourceGate(2)并启动六个delay为10秒的任务。它通过带1秒上限的条件等待确认两个槽位确实被占用后再取消不用固定sleep猜测任务是否开始。然后await批次抛出CancelledError断言active为0。接着先在同一个gate上启动两个holder。每个holder拿到槽位后设置自己的Event然后共同等待release事件因此第一个holder不会提前释放来帮助第二个通过。主任务等两个Event都已设置再断言active为2最后设置release并等待两者退出。这直接证明两个槽位都已归还旧的peak曾经等于2或者下一批能顺序跑完都不能证明这一点。1秒超时只是防挂看门狗实际判据是两个holder同时持有槽位不靠私有字段或性能阈值推断。最后再执行下一批[3,4,5]要求1秒内完成输出值6、8、10ID全部是nextactive仍为0peak不超过2。制作时曾注入一个“取消后少归还一个槽位”的测试替身新holder检查会失败恢复真实实现后通过。这证明检查能够抓到原先可能漏掉的半数容量泄漏。随课队列副本通过显式stopping保留整批正在停止的事实而不是只依赖TaskGroup可能恢复的父任务取消计数。它也保留第10课的取消合同工作项独立取消转WorkCancelled并终止批次外部批次取消仍传播CancelledError。否则消费者悄悄退出时满队列的生产者可能无人唤醒。它与资源门控测试关注的是两个互补问题任务必须收尾已取得的槽位必须成对归还。从课程根目录运行Linux用python3.11替换python。本课核心实验无需供应商服务。python lessons/11/answer_previous.py python lessons/11/main.py python-m unittest discover-s lessons/11-p test_lesson.py-v这套测试不意味着所有Web框架在客户端断开时都会自动取消handler。ASGI应用如何观察断开、路由是否继续工作需要结合框架实际行为验证服务端预算仍应自己实现不能把取消策略寄托在浏览器关闭页面上。E11取消这一批下一批仍能运行原题E11创建一个共享ResourceGate(2)启动6项慢任务并确认active达到2。对批次Task调用cancel后继续await必须收到CancelledError随后active为0。先启动两个holder各自获取同一个gate槽位后设置Event并一起等待release确认两个Event均已设置且active为2再release并await结束以证明两个槽位都归还。然后复用同一个gate运行[3,4,5]在宽松的1秒测试期限内得到[6,8,10]所有行request_id为nextpeak不超过2结束后active为0。调用方事先设置一个request_id整个过程结束后必须恢复该值。不能检查Semaphore私有字段不能把取消吞成正常结果。第12课给完整可运行答案并把资源生命周期接入FastAPI。做到这里判断一个并发函数是否正确已经不只看它返回了什么。还要能说明它创建的任务何时结束、占用的槽位何时归还、调用上下文何时恢复。这些约束会直接决定下一课Web服务在多次请求之间能否稳定工作。取消恰好发生在启动前怎么办如果一个Task刚创建就被取消它的协程正文可能还没开始执行。此时尚未设置请求ID、尚未取得槽位也没有属于它的资源需要释放。测试不能要求每个被创建的任务都必须先增加一次active真正需要保证的是每次成功获取都有一次归还没有获取就不归还。若取消发生在任务已经取得资源但尚未写入结果时finally负责资源清理而结果槽位可以仍然为空因为整个批次会以取消结束调用者不会收到这份半成品。这比在取消时补一个假成功值更明确。取消的业务语义是调用没有完成而不是某个数值恰好等于默认值。上下文恢复也遵循相同原则。设置之前被取消没有token需要reset设置之后的所有退出路径都经过finally。把资源取得与清理放在同一小段结构里让正确性来自控制流本身通常比在函数结尾零散补几行清理语句可靠。测试中的请求ID使用不可变字符串资源门控则刻意由多个批次共享。前者需要上下文隔离后者需要统一容量这两种对象的生命周期不能混为一谈。本课完整源码main.py是资源和上下文实验pipeline.py仅供E10答案复用。测试故意覆盖外部取消与并发上下文不用未来项目才能证明本课结论。lessons/11/answer_previous.pyE10: maxsize2 and completion-independent result positions.importasynciofrompipelineimportrun_queueasyncdefsolve():values[5,4,3,2,1,0]asyncdefwork(value):awaitasyncio.sleep(value*0.005)returnvalue*value results,statsawaitrun_queue(values,work,capacity2,workers2)assertresults[25,16,9,4,1,0]assertstats[max_queued]2andstats[completed]6return{results:results,stats:stats}if__name____main__:print(asyncio.run(solve()))lessons/11/main.pyimportasynciofromcontextlibimportasynccontextmanagerfromcontextvarsimportContextVarimportjson request_idContextVar(request_id,default-)classResourceGate:多个批次共享的资源容量统计值用于观察信号量负责真正限流。def__init__(self,limit):iflimit1:raiseValueError(limit must be positive)self.semaphoreasyncio.Semaphore(limit)self.active0self.peak0asynccontextmanagerasyncdefslot(self):仅成功获取后才计数finally使工作中取消也能归还已取得的槽位。asyncwithself.semaphore:self.active1self.peakmax(self.peak,self.active)try:yieldfinally:self.active-1asyncdefrun_batch(items,gate,batch_id,delay0.03):用可取消等待演示资源生命周期这里只适合有限小批次。 信号量限制活跃工作不限制已创建Task数量大输入应接第10课有界队列。 倍增值仅是实验输出真实报价对应槽位内的HTTP调用和响应处理。 tokenrequest_id.set(batch_id)results[None]*len(items)try:asyncdefone(index,item):asyncwithgate.slot():awaitasyncio.sleep(delay)results[index]{value:item*2,request_id:request_id.get()}asyncwithasyncio.TaskGroup()asgroup:forindex,iteminenumerate(items):group.create_task(one(index,item))returnresultsfinally:# 恢复调用前的上下文不能写死默认值取消和异常也必须执行。request_id.reset(token)asyncdefmain():gateResourceGate(2)rowsawaitrun_batch([1,2,3],gate,demo)print(json.dumps({results:rows,active_after:gate.active,peak:gate.peak,caller_request_id:request_id.get()},indent2))if__name____main__:asyncio.run(main())lessons/11/pipeline.pyimportasyncioclassWorkCancelled(RuntimeError):工作项单独取消导致结果缺失应让整个队列失败而不是丢掉消费者。asyncdefrun_queue(items,work,capacity2,workers2):有界输入和固定工作者工作项自取消失败即停外部批次取消继续传播。ifcapacity1orworkers1:raiseValueError(capacity and workers must be positive)queueasyncio.Queue(maxsizecapacity)sentinelobject()results[]stats{max_queued:0,completed:0}ownerasyncio.current_task()stoppingFalseasyncdefproduce():nonlocalstopping# 满队列让生产者暂停索引随任务旅行结果不依赖完成顺序。try:foriteminitems:indexlen(results)results.append(None)awaitqueue.put((index,item))stats[max_queued]max(stats[max_queued],queue.qsize())for_inrange(workers):awaitqueue.put(sentinel)exceptException:# 先标记整批收尾再让TaskGroup取消兄弟任务。stoppingTrueraiseasyncdefconsume():nonlocalstoppingwhileTrue:entryawaitqueue.get()try:ifentryissentinel:returnindex,itementrytry:results[index]awaitwork(item)exceptasyncio.CancelledErrorasexc:# stopping保存批次语义不依赖TaskGroup暂时增减的取消计数。# owner计数仅帮助识别尚未被主任务捕获的外部取消。ifstoppingorowner.cancelling():stoppingTrueraisestoppingTrueraiseWorkCancelled(fwork item{index}cancelled independently)fromexc stats[completed]1exceptException:stoppingTrueraisefinally:# 只确认已经get的条目包括哨兵失败由任务组负责传播。queue.task_done()asyncwithasyncio.TaskGroup()asgroup:try:producergroup.create_task(produce())for_inrange(workers):group.create_task(consume())# 先确认不会再有put才能用join判断整批任务均已确认。awaitproducerawaitqueue.join()exceptasyncio.CancelledError:# 在任务组开始异步收尾前记住批次取消继续向上传播。stoppingTrueraisereturnresults,statslessons/11/test_lesson.pyimportasyncioimportunittestfrommainimportResourceGate,request_id,run_batchfrompipelineimportWorkCancelled,run_queueclassLessonTests(unittest.IsolatedAsyncioTestCase):asyncdeftest_copied_pipeline_reports_independent_work_cancellation(self):asyncdefwork(value):cancelledasyncio.get_running_loop().create_future()cancelled.cancel()awaitcancelledwithself.assertRaises(ExceptionGroup)ascaught:asyncwithasyncio.timeout(1):awaitrun_queue(range(100),work,capacity2,workers2)self.assertTrue(all(isinstance(exc,WorkCancelled)forexcincaught.exception.exceptions))asyncdeftest_nested_call_restores_existing_outer_context(self):tokenrequest_id.set(outer)try:rowsawaitrun_batch([1],ResourceGate(1),inner)self.assertEqual(rows[0][request_id],inner)self.assertEqual(request_id.get(),outer)finally:request_id.reset(token)asyncdeftest_external_cancel_releases_slots_and_next_batch_works(self):gateResourceGate(2)taskasyncio.create_task(run_batch(range(6),gate,cancelled,delay10))asyncwithasyncio.timeout(1):whilegate.active2:awaitasyncio.sleep(0)task.cancel()withself.assertRaises(asyncio.CancelledError):awaittask self.assertEqual(gate.active,0)# 两个holder在同一时刻保留槽位顺序跑完下一批不能证明两个都归还。entered[asyncio.Event(),asyncio.Event()]releaseasyncio.Event()asyncdefhold(index):asyncwithgate.slot():entered[index].set()awaitrelease.wait()asyncwithasyncio.TaskGroup()asgroup:forindexinrange(2):group.create_task(hold(index))try:asyncwithasyncio.timeout(1):awaitasyncio.gather(*(event.wait()foreventinentered))self.assertEqual(gate.active,2)finally:release.set()self.assertEqual(gate.active,0)asyncwithasyncio.timeout(1):rowsawaitrun_batch([3,4,5],gate,next,delay0.01)self.assertEqual([row[value]forrowinrows],[6,8,10])self.assertTrue(all(row[request_id]nextforrowinrows))self.assertEqual(gate.active,0)self.assertLessEqual(gate.peak,2)self.assertEqual(request_id.get(),-)asyncdeftest_concurrent_batches_do_not_share_context_values(self):gateResourceGate(2)left,rightawaitasyncio.gather(run_batch([1,2],gate,A),run_batch([3,4],gate,B))self.assertEqual({row[request_id]forrowinleft},{A})self.assertEqual({row[request_id]forrowinright},{B})self.assertEqual(request_id.get(),-)if__name____main__:unittest.main()