python - 在两个不同的线程中运行 asyncio 以动态处理操作

Question

我正在尝试异步处理许多操作:我想将操作发送到我的循环并同时在 ProcessPoolExecutor 中运行它们。我想我不知道一开始将运行的所有作业，因此我无法定义所有作业并然后启动事件循环。

我发现的唯一解决方案是运行一个可以处理操作的主线程，以及另一个线程来执行loop.run_forever，这似乎可行。但是，我没有看到任何两个单独的线程以这种方式在同一循环上运行的示例。有没有其他方法可以解决这个问题，如果我的解决方案会出现什么问题？

import asyncio
from concurrent.futures import ProcessPoolExecutor
import functools
import time
import threading


executor = ProcessPoolExecutor(max_workers=3)


def do_work(eventloop, value):
    future = eventloop.run_in_executor(executor, functools.partial(process_action, value))
    future.add_done_callback(run_job_success)


def process_action(value):
    print("Processing %i" % value)
    time.sleep(1)
    return value


def run_job_success(f):
    print("Success : %s" % f.result())


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop_thread = threading.Thread(target=loop.run_forever)
    loop_thread.start()
    while True:
        msg = recv()
        if msg is not None:
            do_work(loop, msg)

编辑: 我使用 recv 方法获取要接收的作业。

Answer 1

您想要做的事情有点含糊 - 您说您不知道在程序开始时要运行的所有作业。没问题 - 但您如何找到您想要从事的工作呢？无论如何，您可以(并且应该)使用 BaseEventLoop.call_soon 来将上面的测试程序重写为单线程，以安排您想要的所有 do_work 调用在启动事件循环之前进行:

import asyncio
from concurrent.futures import ProcessPoolExecutor
import functools
import time

def do_work(eventloop, value):
    future = eventloop.run_in_executor(executor, functools.partial(process_action, value))
    future.add_done_callback(run_job_success)

def process_action(value):
    print("Processing %i" % value)
    time.sleep(1)
    return value

def run_job_success(f):
    print("Success : %s" % f.result())

if __name__ == "__main__":
    executor = ProcessPoolExecutor(max_workers=3)
    loop = asyncio.get_event_loop()
    for i in range(5):
        loop.call_soon(do_work, loop, i)
    loop.run_forever()

或者可以进一步重构，使用协程而不是回调，这通常是使用 asyncio 时的首选样式:

import time
import asyncio
import functools
from concurrent.futures import ProcessPoolExecutor

def do_work(loop, value):
    return loop.run_in_executor(executor, functools.partial(process_action, value))

def process_action(value):
    print("Processing %i" % value)
    time.sleep(1)
    return value

@asyncio.coroutine
def main(loop):
    tasks = [do_work(loop, i) for i in range(5)]
    for fut in asyncio.as_completed(tasks):
        result = yield from fut
        print("Success : %s" % result)

if __name__ == "__main__":
    executor = ProcessPoolExecutor(max_workers=3)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

这也使得在所有工作完成后退出程序变得更加容易，而不必使用 Ctrl+C 和 loop.run_forever。

您当前的方法是安全的(loop.run_in_executor 在幕后使用loop.call_soon_threadsafe，这是您可以安全/正确地在事件循环中安排工作的唯一方法来自一个单独的线程)，它只是过于复杂且不必要； asyncio 的设计使得使用它的程序是单线程的(除非需要运行阻塞操作，这就是 run_in_executor 的用途)。