异步等待主进程中的多处理队列

Question

我有以下情形：多个工作进程将有关其当前状态的事件发送到事件调度程序。然后，如果我们在主流程中，则此事件分发程序需要处理所有事件；如果在工作进程中，则该信号将需要主进程的事件分发程序发信号以处理这些事件。

这里的主要问题是事件处理也必须在主进程的主线程中，所以我不能只在线程中运行一会儿True循环并在那里等待工作进程的消息。

所以我有这个：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import current_process, Process, Queue
from threading import current_thread
from time import sleep

def get_q(q):
    print("Waiting for the queue ({} / {})\n".format(current_thread().name, current_process().name))
    return q.get()

async def message_q(q):
    while True:
        f = loop.run_in_executor(None, get_q, q)

        await f

        if f.result() is None:
            print("Done")
            return;

        print("Got the result ({} / {})".format(current_thread().name, current_process().name))
        print("Result is: {}\n".format(f.result()))

async def something_else():
    while True:
        print("Something else\n")
        await asyncio.sleep(2)

def other_process(q):
    for i in range(5):
        print("Putting something in the queue ({})".format(current_process().name))
        q.put(i)
        sleep(1)

    q.put(None)

q = Queue()

Process(target=other_process, args=(q,), daemon=True).start()

loop = asyncio.get_event_loop()
loop.set_default_executor(ThreadPoolExecutor(max_workers=1))
asyncio.ensure_future(message_q(q))
asyncio.ensure_future(something_else())
loop.run_until_complete(asyncio.sleep(6))

other_process() is an exemplary worker process which uses a Queue to signal the main process which runs an event loop to process stuff and also wait for any data on the Queue. In the real case, this process would signal the event dispatcher which would then handle the queue messaging, to pass the message on to the main process event dispatcher, but here I simplified it a bit.

However, I'm not quite satisfied with this. Submitting get_q() again and again to a ThreadPoolExecutor produces more overhead and isn't as clean as just one long-running thread. Also the await f there isn't optimal and blocks as soon as no further data is in the queue, which prevents the event loop from exiting. My workaround is to send None after the workers have finished and exit message_q() if None is in the queue.

有没有更好的方法来实现这一目标？性能非常关键，我想将Queue对象保留在事件分发程序的本地，而不要将其传递给管理工作进程的代码（或要求调用finalize()其某种方法）。

Answer 1

我现在将其实现为异步上下文管理器。上下文管理器调用

asyncio.ensure_future(message_q())

在其__aenter__()方法中，并在其方法中添加None到队列中__aexit__()，以关闭 中的无限循环message_q()。

async with然后，可以在围绕进程生成代码部分的语句中使用上下文管理器，从而无需手动调用关闭方法。但是，建议在确保协程允许上下文管理器初始化队列侦听器后在方法await asyncio.sleep(0)内部调用。否则，不会立即被调用。这本身不是问题（因为队列无论如何都已满），但它会延迟事件转发，直到代码中发生下一个事件。__aenter__()message_q()message_q()await

这些进程应该使用 和ProcessPoolExecutor一起生成loop.run_in_executor()，因此等待进程不会阻塞事件循环。

除了使用队列之外，您可能还想使用 JoinableQueue 来确保在退出上下文管理器之前所有事件都已得到处理。