Python - 将多处理与 asyncio 结合起来有时才有效

Question

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

代码是这样的，我有一个具有一些非阻塞函数和一个阻塞的类。我有一个 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")

大约有一半的时间我运行它，它工作正常并打印出阻塞和非阻塞队列中的文本。另一半只打印出非阻塞队列的结果。看起来阻塞过程根本没有开始。并非每隔一段时间都会发生这种情况。它可能连续工作五次，然后连续工作五次。

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

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/任务内部有一些状态需要保持一致或在另一个进程中运行时被破坏