Python

Question

我想结合 asyncio 和多处理，因为我有一个任务，其中一部分是 io-bound，另一个是 cpu-bound。我首先尝试使用 loop.run_in_executor()，但我可能无法让它工作。相反，我创建了两个进程，一个使用 asyncio，另一个不使用。

代码是这样的，我有一个 class 具有一些非阻塞函数和一个阻塞函数。我有一个 asyncio.Queue 在非阻塞部分之间传递信息，还有一个 multiprocessing.Queue 在非阻塞和阻塞函数之间传递信息。

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
import asyncio
import time


class TestClass:
    def __init__(self):
        m = mp.Manager()
        self.blocking_queue = m.Queue()

    async def run(self):
        loop = asyncio.get_event_loop()
        self.non_blocking_queue = asyncio.Queue() # asyncio Queue must be declared within event loop
        task1 = loop.create_task(self.non_blocking1())
        task2 = loop.create_task(self.non_blocking2())
        task3 = loop.create_task(self.print_msgs())
        await asyncio.gather(task1, task2)
        task3.cancel()

    def blocking(self):
        i = 0
        while i < 5:
            time.sleep(0.6)
            i += 1
            print("Blocking ", i)
            line = self.blocking_queue.get()
            print("Blocking: ", line)
        print("blocking done")

    async def non_blocking1(self):
        for i in range(5):
            await self.non_blocking_queue.put("Hello")
            await asyncio.sleep(0.4)

    async def non_blocking2(self):
        for i in range(5):
            await self.non_blocking_queue.put("World")
            await asyncio.sleep(0.5)

    async def print_msgs(self):
        while True:
            line = await self.non_blocking_queue.get()
            self.blocking_queue.put(line)
            print(line)


test_class = TestClass()
with ProcessPoolExecutor() as pool:
    pool.submit(test_class.blocking)
    pool.submit(asyncio.run(test_class.run()))
print("done")

大约一半的时间我运行这个，它工作正常并打印出阻塞和非阻塞队列中的文本。另一半它只打印非阻塞队列的结果。看起来阻塞过程根本没有开始。不是每次都这样。可能连续五次有效，然后连续五次无效。

什么可能导致这样的问题？同时使用多处理和 asyncio，我可以用哪种更好的方法来做到这一点？

Answer 1

运行另一个进程“内部”的异步任务对我有用，例如：

def runfn(fn):
    return asyncio.run(fn())

with ProcessPoolExecutor() as pool:
    pool.submit(test_class.blocking)
    pool.submit(runfn, test_class.run)

大概 asyncio/the 任务中有一些状态需要保持一致，或者当运行在另一个进程中时被破坏

Python - 将多处理与 asyncio 结合使用仅在某些时候有效

Python - Combining multiprocessing with asyncio works only sometimes

multiprocessing

python-asyncio

python-multiprocessing