如何实现兼具手动操作与with语句自动管理的对象?以asyncio.Lock为例
让类同时支持手动操作与异步上下文管理(以自定义asyncio锁为例)
你提到的open函数的两种用法,本质是返回的文件对象同时实现了手动调用方法(readline()/close())和同步上下文管理器(__enter__/__exit__)。对于异步场景的自定义锁,我们只需要给类加上异步上下文管理器的魔法方法,就能同时支持手动acquire/release和async with自动管理。
自定义异步锁实现代码
import asyncio class CustomAsyncLock: def __init__(self): # 直接复用原生asyncio.Lock的核心逻辑,避免重复造轮子 self._inner_lock = asyncio.Lock() # 手动获取锁的方法 async def acquire(self): await self._inner_lock.acquire() # 手动释放锁的方法 def release(self): self._inner_lock.release() # 异步上下文管理器:进入代码块时自动获取锁 async def __aenter__(self): await self.acquire() return self # 异步上下文管理器:退出代码块时自动释放锁 async def __aexit__(self, exc_type, exc_val, exc_tb): self.release() # 异常会自动向外传播,这里不需要额外处理
两种使用方式示例
1. 手动管理锁的获取与释放
async def manual_operation(): lock = CustomAsyncLock() await lock.acquire() try: print("手动拿到锁,执行需要加锁的异步操作") finally: # 必须手动释放,避免死锁 lock.release() asyncio.run(manual_operation())
2. 用async with自动管理锁
async def auto_operation(): lock = CustomAsyncLock() async with lock: print("用async with自动获取锁,代码块结束后自动释放") asyncio.run(auto_operation())
关键说明
async with语法会自动触发__aenter__方法获取锁,代码块执行完毕(无论是否抛出异常)都会调用__aexit__释放锁,省去了手动编写try-finally的冗余代码- 保留
acquire和release方法,让你在需要精细控制锁的生命周期场景下,依然可以手动操作
内容的提问来源于stack exchange,提问作者Rillian Grant
相关产品推荐
相关产品推荐

