asyncio无RLock时，避免重复加锁死锁的实现方案咨询

一、能否检查当前任务是否持有指定锁？ #

asyncio 内置的Lock没有直接提供检查当前任务是否持有锁的公开API——它的设计偏向轻量，内部没有记录锁的持有者信息。但我们可以通过自定义扩展的方式，给锁添加持有者追踪逻辑，实现这个检查能力。

二、记录锁状态避免重复加锁的最佳实践 #

1. 自定义可重入异步锁（AsyncRLock） #

这是最直接的解决方案，完全模拟线程中RLock的逻辑，给锁加上持有者追踪和重入计数：

import asyncio
from typing import Optional

class AsyncRLock:
    def __init__(self):
        self._lock = asyncio.Lock()
        self._owner: Optional[asyncio.Task] = None
        self._reentrant_count = 0

    async def acquire(self):
        current_task = asyncio.current_task()
        # 如果当前任务已经持有锁，直接增加重入计数
        if self._owner == current_task:
            self._reentrant_count += 1
            return
        # 否则等待获取锁，记录持有者
        await self._lock.acquire()
        self._owner = current_task
        self._reentrant_count = 1

    def release(self):
        current_task = asyncio.current_task()
        if self._owner != current_task:
            raise RuntimeError("无法释放当前任务未持有的锁")
        self._reentrant_count -= 1
        # 重入计数归0时，真正释放底层锁并清空持有者记录
        if self._reentrant_count == 0:
            self._owner = None
            self._lock.release()

    # 支持async with语法
    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self.release()

使用方式和普通Lock完全一致，同一任务多次获取不会死锁：

lock = AsyncRLock()

async def core_logic():
    async with lock:
        print("首次获取锁，执行核心逻辑")
        await handle_exception()

async def handle_exception():
    async with lock:
        print("同一任务重入锁，执行异常处理逻辑")

asyncio.run(core_logic())

2. 手动追踪锁持有状态（适合简单场景） #

如果不想自定义锁，可以利用任务的局部属性记录锁的持有状态：

import asyncio

lock = asyncio.Lock()

async def safe_acquire(target_lock):
    current_task = asyncio.current_task()
    # 用锁的ID作为属性后缀，避免不同锁的状态冲突
    lock_flag = f"_hold_lock_{id(target_lock)}"
    if hasattr(current_task, lock_flag):
        return
    await target_lock.acquire()
    setattr(current_task, lock_flag, True)

def safe_release(target_lock):
    current_task = asyncio.current_task()
    lock_flag = f"_hold_lock_{id(target_lock)}"
    if not hasattr(current_task, lock_flag):
        raise RuntimeError("当前任务未持有该锁")
    delattr(current_task, lock_flag)
    target_lock.release()

# 使用示例
async def core_logic():
    await safe_acquire(lock)
    try:
        print("执行核心逻辑")
        await handle_exception()
    finally:
        safe_release(lock)

async def handle_exception():
    await safe_acquire(lock)
    try:
        print("执行异常处理")
    finally:
        safe_release(lock)

这种方式要注意：必须在finally块里释放锁并清理任务属性，避免锁泄漏。

3. 用第三方成熟库 #

如果不想自己实现，可以直接用asyncio-extras这类库中已经实现好的可重入锁，它们已经处理了任务取消、状态清理等边缘情况。

三、关键注意事项 #

• 自定义AsyncRLock时，要处理任务取消的场景：如果任务在持有锁时被取消，要确保锁状态能正确重置，可结合asyncio.shield或在__aexit__中做兜底处理。
• 不要依赖Lock的私有属性（比如_locked）来判断状态，这类API没有公开承诺，版本更新可能会失效。
• 手动追踪状态时，属性命名要足够独特，避免和任务其他属性冲突。

内容的提问来源于stack exchange，提问作者tamirg