如何将 asyncio 与其他操作系统线程同步？

Question

我有一个带有一个主线程的程序，我在其中生成了第二个使用 asyncio 的线程。是否提供了任何工具来同步这两个线程？如果一切都是异步的，我可以用它的同步原语来做，例如：

import asyncio

async def taskA(lst, evt):
    print(f'Appending 1')
    lst.append(1)
    evt.set()

async def taskB(lst, evt):
    await evt.wait()
    print('Retrieved:', lst.pop())

lst = []
evt = asyncio.Event()
asyncio.get_event_loop().run_until_complete(asyncio.gather(
    taskA(lst, evt),
    taskB(lst, evt),
))

但是，这不适用于多线程。如果我只使用threading.Event，那么它将阻塞异步线程。我想我可以将等待推迟到执行人：

import asyncio
import threading

def taskA(lst, evt):
    print(f'Appending 1')
    lst.append(1)
    evt.set()

async def taskB(lst, evt):
    asyncio.get_event_loop().run_in_executor(None, evt.wait)
    print('Retrieved:', lst.pop())

def targetA(lst, evt):
    taskA(lst, evt)

def targetB(lst, evt):
    asyncio.set_event_loop(asyncio.new_event_loop())
    asyncio.get_event_loop().run_until_complete(taskB(lst, evt))

lst = []
evt = threading.Event()
threadA = threading.Thread(target=targetA, args=(lst, evt))
threadB = threading.Thread(target=targetB, args=(lst, evt))
threadA.start()
threadB.start()
threadA.join()
threadB.join()

但是，让执行线程只等待互斥锁似乎不自然。这是应该这样做的方式吗？或者有没有其他方法可以异步等待操作系统线程之间的同步？

Answer 1

将 asyncio 协程与来自另一个线程的事件同步的一种简单方法是在 taskB 中等待 asyncio.Event，然后使用 loop.call_soon_threadsafe 从 taskA 设置它。

为了能够在两者之间传递值和异常，可以使用futures；但是，您正在发明run_in_executor 的大部分内容。如果 taskA 的唯一工作是将任务从队列中取出，那么您不妨创建一个单工作“池”并将其用作工作线程。然后您可以按预期使用run_in_executor：

worker = concurrent.futures.ThreadPoolExecutor(max_workers=1)

async def taskB(lst):
    loop = asyncio.get_event_loop()
    # or result = await ..., if taskA has a useful return value
    # This will also propagate exceptions raised by taskA
    await loop.run_in_executor(worker, taskA, lst)
    print('Retrieved:', lst.pop())

语义与带有显式队列的版本相同 - 队列仍然存在，它就在 ThreadPoolExecutor 内。