Python asyncio run_coroutine_threadsafe 从不运行协程？

Question

我不确定我在这里做错了什么，我试图拥有一个包含队列并使用协程来消耗该队列上的项目的类。问题是事件循环正在一个单独的线程中运行（在该线程中我是为了loop.run_forever()让它运行）。

但我看到的是，用于消费项目的协程从未被触发：

import asyncio
from threading import Thread

import functools

# so print always flushes to stdout
print = functools.partial(print, flush=True)


def start_loop(loop):
    def run_forever(loop):
        print("Setting loop to run forever")
        asyncio.set_event_loop(loop)
        loop.run_forever()
        print("Leaving run forever")

    asyncio.set_event_loop(loop)
    print("Spawaning thread")
    thread = Thread(target=run_forever, args=(loop,))
    thread.start()


class Foo:
    def __init__(self, loop):
        print("in foo init")
        self.queue = asyncio.Queue()
        asyncio.run_coroutine_threadsafe(self.consumer(self.queue), loop)

    async def consumer(self, queue):
        print("In consumer")
        while True:
            message = await queue.get()
            print(f"Got message {message}")
            if message == "END OF QUEUE":
                print(f"exiting consumer")
                break
            print(f"Processing {message}...")


def main():
    loop = asyncio.new_event_loop()
    start_loop(loop)
    f = Foo(loop)
    f.queue.put("this is a message")
    f.queue.put("END OF QUEUE")

    loop.call_soon_threadsafe(loop.stop)

    # wait for the stop to propagate and complete
    while loop.is_running():
        pass


if __name__ == "__main__":
    main()

输出：

产卵线程
设置循环永远运行
在 foo 初始化中
永远离开跑步

Answer 1

这段代码有几个问题。

首先，检查警告：

test.py:44: RuntimeWarning: coroutine 'Queue.put' was never awaited
f.queue.put("this is a message")
test.py:45: RuntimeWarning: coroutine 'Queue.put' was never awaited
f.queue.put("END OF QUEUE")

这意味着queue.put是一个协程，所以它必须使用run_coroutine_threadsafe运行：

asyncio.run_coroutine_threadsafe(f.queue.put("this is a message"), loop)
asyncio.run_coroutine_threadsafe(f.queue.put("END OF QUEUE"), loop)

您还可以使用queue.put_nowait，这是一种同步方法。然而，asyncio 对象通常不是线程安全的，因此每个同步调用都必须经过call_soon_threadsafe：

loop.call_soon_threadsafe(f.queue.put_nowait, "this is a message")
loop.call_soon_threadsafe(f.queue.put_nowait, "END OF QUEUE")

另一个问题是循环在消费者任务开始处理项目之前停止。您可以join向类添加一个方法Foo来等待使用者完成：

class Foo:
    def __init__(self, loop):
        [...]
        self.future = asyncio.run_coroutine_threadsafe(self.consumer(self.queue), loop)

    def join(self):
        self.future.result()

然后确保在停止循环之前调用此方法：

f.join()
loop.call_soon_threadsafe(loop.stop)

这应该足以让程序按您的预期工作。然而，这段代码在几个方面仍然存在问题。

首先，循环不应该同时设置在主线程和额外线程中。异步循环并不意味着在线程之间共享，因此您需要确保与异步相关的所有内容都发生在专用线程中。

由于Foo负责这两个线程之间的通信，因此您必须格外小心，以确保每行代码都在正确的线程中运行。例如，实例化asyncio.Queue必须发生在 asyncio 线程中。

请参阅此要点以获取程序的更正版本。

另外，我想指出这不是 asyncio 的典型用例。您通常希望在主线程中运行异步循环，特别是如果您需要子进程支持：

asyncio 支持从不同线程运行子进程，但有限制：

• 事件循环必须在主线程中运行
• 在从其他线程执行子进程之前，必须在主线程中实例化子观察程序。在主线程中调用 get_child_watcher() 函数来实例化子观察者。

我建议以另一种方式设计您的应用程序，即在主线程中运行 asyncio 并使用run_in_executor作为同步阻塞代码。