You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何实现兼具手动操作与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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.11 13:15:56