如何实现可处理未知步骤且能保证优雅退出上下文管理器的Python协程?
解决协程中上下文管理器自动退出的问题
你的问题核心在于:基于生成器的协程会一直挂在while True的yield处,上下文管理器的退出逻辑(yield之后的代码)只有当with块正常结束或遇到未捕获的异常时才会执行,而默认情况下,Python不会自动关闭挂起的生成器,所以上下文永远不会退出。
要实现协程被回收或超出作用域时自动退出上下文,我们可以利用生成器的close()机制——当生成器对象被关闭(手动调用close()或被垃圾回收时自动调用),Python会在生成器挂起的yield处抛出GeneratorExit异常,我们可以通过捕获这个异常让协程退出循环,进而触发上下文管理器的清理逻辑。
方案1:在协程中捕获GeneratorExit并配合上下文管理器的finally
首先修改上下文管理器,确保清理逻辑在finally块中(即使有异常也能执行),然后在协程的循环中捕获GeneratorExit:
import contextlib @contextlib.contextmanager def cm(): print("STARTED") try: yield finally: # 把清理逻辑放在finally里,确保无论如何都会执行 print("ENDED") def coro(a: str): with cm(): print(a) try: while True: val1, val2 = yield print(val1, val2) except GeneratorExit: # 捕获GeneratorExit,让协程正常退出循环,触发with块的结束 pass
测试代码
如果手动调用close(),会立刻触发上下文退出:
c = coro("HI") c.send(None) print("---") c.send((1, 2)) print("---!") # 手动关闭协程 c.close()
输出结果:
STARTED HI --- 1 2 ---! ENDED
如果不想手动调用close(),可以依赖Python的垃圾回收。不过垃圾回收时机不确定,你可以手动触发:
c = coro("HI") c.send(None) c.send((1, 2)) # 删除引用 del c # 手动触发垃圾回收 import gc gc.collect()
这时候也会输出ENDED。
方案2:包装协程为类,利用__del__自动关闭
如果你不想让调用方关心close()的调用,可以把协程包装成一个类,在__del__方法中自动调用生成器的close():
import contextlib import gc @contextlib.contextmanager def cm(): print("STARTED") try: yield finally: print("ENDED") def coro(a: str): with cm(): print(a) try: while True: val1, val2 = yield print(val1, val2) except GeneratorExit: pass class ManagedCoro: def __init__(self, a): self._gen = coro(a) def send(self, value): return self._gen.send(value) def __del__(self): try: self._gen.close() except Exception: # 忽略关闭时可能出现的异常 pass # 使用示例 c = ManagedCoro("HI") c.send(None) c.send((1, 2)) # 当对象超出作用域或被删除时,__del__会自动调用close() del c gc.collect()
这个方案对调用方更友好,不需要手动处理关闭逻辑,只要协程对象被回收,上下文就会自动退出。
原理说明
- 生成器的
close()方法会向挂起的生成器抛出GeneratorExit异常,生成器函数会从yield处退出,进而执行with块的退出逻辑(上下文管理器的__exit__方法,对应contextmanager装饰器中yield之后的代码)。 - 当生成器对象被垃圾回收时,Python会自动调用其
close()方法,所以只要没有循环引用导致垃圾回收无法触发,上下文最终会被清理。
内容的提问来源于stack exchange,提问作者ikamen
相关产品推荐
相关产品推荐

