python中异步任务和同步线程之间的通信答案

【问题标题】：Communication between async tasks and synchronous threads in pythonpython中异步任务和同步线程之间的通信
【发布时间】：2020-01-08 16:38:14
【问题描述】：

我正在寻找异步任务和方法/函数之间通信的最佳解决方案，这些方法/函数在 concurrent.futures 的线程池执行器中运行。在之前的同步项目中，我会使用queue.Queue 类。我认为任何方法都应该是线程安全的，因此 asyncio.queue 将不起作用。

我见过人们扩展 queue.Queue 类来做类似的事情：

class async_queue(Queue):
  async def aput(self, item):
    self.put_nowait(item)

  async def aget(self):
    resp = await asyncio.get_event_loop().run_in_executor( None, self.get )
    return resp

有没有更好的方法？

【问题讨论】：

标签： python python-asyncio

【解决方案1】：

我建议反过来：使用asyncio.Queue 类在两个世界之间进行通信。这样做的好处是不必将线程池中的一个槽用于需要很长时间才能完成的操作，例如get()。

这是一个例子：

class Queue:
    def __init__(self):
        self._loop = asyncio.get_running_loop()
        self._queue = asyncio.Queue()

    def sync_put_nowait(self, item):
        self._loop.call_soon(self._queue.put_nowait, item)

    def sync_put(self, item):
        asyncio.run_coroutine_threadsafe(self._queue.put(item), self._loop).result()

    def sync_get(self):
        return asyncio.run_coroutine_threadsafe(self._queue.get(item), self._loop).result()

    def async_put_nowait(self, item):
        self._queue.put_nowait(item)

    async def async_put(self, item):
        await self._queue.put(item)

    async def async_get(self):
        return await self._queue.get()

以sync_ 为前缀的方法旨在由同步代码调用（在事件循环线程之外运行）。以async_ 为前缀的那些将由在事件循环线程中运行的代码调用，无论它们是否实际上是协程。（例如put_nowait 不是协程，但它仍然必须区分同步版本和异步版本。）

【讨论】：