如何在异步函数中使用 threading.Lock 而可以从多个线程访问对象

Question

我想threading.Lock()在异步函数中使用，asyncio.Lock()不是线程安全的，所以我不能这样做with await asyncio.Lock():。我需要使用的原因threading.Lock()是因为这个对象可能被多个线程访问，因此它被用在一个网络应用程序中，运行它的服务器可以启动许多线程。这样做的有效方法是什么？到目前为止，我已经尝试了一个使用锁的简单函数：1)async def main():    with await threading.Lock():        a = 6    return aTypeError: object _thread.lock can't be used in 'await' expressionasync def main():    async with threading.Lock():            a = 1564666546    return aAttributeError: __aexit__

Answer 1

您不能将 a 传递threading.Lock给async with，因为它不是为异步使用而设计的，它是一个阻塞原语。更重要的是，async with threading.Lock()即使它确实有效也没有意义，因为您将获得一把全新的锁，它总是会成功。为了使锁定有意义，您必须在多个线程之间共享一个锁，例如存储在对象的属性中，或以另一种方式与对象相关联。这个答案的其余部分将假设您在threading.Lock线程之间共享。

由于threading.Lock总是阻塞，因此您可以从 asyncio 使用它的唯一方法是在专用线程中获取它，暂停当前协程的执行，直到获取锁。此功能已包含在run_in_executor事件循环方法中，您可以应用该方法：

_pool = concurrent.futures.ThreadPoolExecutor()

async def work(lock, other_args...):

    # lock is a threading.Lock shared between threads

    loop = asyncio.get_event_loop()

    # Acquire the lock in a worker thread, suspending us while waiting.

    await loop.run_in_executor(_pool, lock.acquire)

    ... access the object with the lock held ...

    # Can release directly because release() doesn't block and a

    # threading.Lock can be released from any thread.

    lock.release()

您可以通过创建异步上下文管理器来使其使用起来更优雅（并且异常安全）：

_pool = concurrent.futures.ThreadPoolExecutor()

@contextlib.asynccontextmanager

async def async_lock(lock):

    loop = asyncio.get_event_loop()

    await loop.run_in_executor(_pool, lock.acquire)

    try:

        yield  # the lock is held

    finally:

        lock.release()

然后你可以按如下方式使用它：

# lock is a threading.Lock shared between threads

async with async_lock(lock):

    ... access the object with the lock held ...

当然，在 asyncio 之外你不会使用其中的任何一个，你只是直接获取锁：

# lock is a threading.Lock shared between threads

with lock:

   ... access the object ...

请注意，我们使用单独的线程池而不是传递None给run_in_executor()重用默认池。这是为了避免在持有锁的函数本身需要访问线程池以供其他用途的情况下出现死锁run_in_executor()。通过保持线程池私有，我们避免了因其他人使用同一池而导致死锁的可能性。