有没有一种方法可以在多个线程中使用asyncio.Queue？

Question

假设我有以下代码：

import asyncio
import threading

queue = asyncio.Queue()

def threaded():
    import time
    while True:
        time.sleep(2)
        queue.put_nowait(time.time())
        print(queue.qsize())

@asyncio.coroutine
def async():
    while True:
        time = yield from queue.get()
        print(time)

loop = asyncio.get_event_loop()
asyncio.Task(async())
threading.Thread(target=threaded).start()
loop.run_forever()

这段代码的问题在于，async协程内部的循环永远不会完成第一次迭代，而queue大小会不断增加。

为什么会这样发生，我该怎么解决？

我无法摆脱单独的线程，因为在我的真实代码中，我使用了单独的线程与串行设备进行通信，而且我还没有找到使用的方法asyncio。

Answer 1

asyncio.Queue 不是线程安全的，因此您不能直接在多个线程中直接使用它。相反，您可以使用janus，它是提供线程感知asyncio队列的第三方库：

import asyncio
import threading
import janus

def threaded(squeue):
    import time
    while True:
        time.sleep(2)
        squeue.put_nowait(time.time())
        print(squeue.qsize())

@asyncio.coroutine
def async(aqueue):
    while True:
        time = yield from aqueue.get()
        print(time)

loop = asyncio.get_event_loop()
queue = janus.Queue(loop=loop)
asyncio.Task(asyncio.ensure_future(queue.async_q))
threading.Thread(target=threaded, args=(queue.sync_q,)).start()
loop.run_forever()

还有aioprocessing（全披露：我写了它），它还提供了进程安全的队列（作为副作用，线程安全的队列），但是如果您不尝试使用，那就太过分了multiprocessing。

编辑

正如在其他答案中指出的那样，对于简单的用例，您也可以使用loop.call_soon_threadsafe添加到队列中。