异步 + 多处理 + unix

Question

我有一个具有以下逻辑的宠物项目：

import asyncio, multiprocessing

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())

def start():
    multiprocessing.Process(target=sub_loop).start()

start()

如果你运行它，你会看到：

Hello from subprocess

这很好。但我要做的是改为使start()协程：

async def start():
    multiprocessing.Process(target=sub_loop).start()

要运行它，我必须这样做：

asyncio.get_event_loop().run_until_complete(start())

问题是：创建子进程时，它会克隆整个 Python 环境，因此事件循环已经在那里运行：

Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "test.py", line 7, in sub_loop
    asyncio.get_event_loop().run_until_complete(sub_main())
  File "/usr/lib/python3.5/asyncio/base_events.py", line 361, in run_until_complete
    self.run_forever()
  File "/usr/lib/python3.5/asyncio/base_events.py", line 326, in run_forever
    raise RuntimeError('Event loop is running.')
RuntimeError: Event loop is running.

我试图在没有运气的情况下在子进程端销毁它，但我认为正确的方法是防止它与子进程共享。有可能吗？

更新： 这是完整的失败代码：

import asyncio, multiprocessing

import asyncio.unix_events

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())


async def start():
    multiprocessing.Process(target=sub_loop).start()

asyncio.get_event_loop().run_until_complete(start())

Answer 1

您应该始终添加检查以查看您是如何运行代码的（if __name__ == '__main__': 部分。您的子进程第二次运行模块中的所有内容，给您带来悲伤（可能t抵抗）。

import asyncio, multiprocessing

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())


async def start():
    multiprocessing.Process(target=sub_loop).start()

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(start())

Answer 2

首先，如果您计划在循环中运行 python 子进程，您应该考虑使用 loop.run_in_executor 和 ProcessPoolExecutor。至于你的问题，你可以使用event loop policy函数来设置一个新的循环：

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(sub_main())

async def start(executor):
    await asyncio.get_event_loop().run_in_executor(executor, sub_loop)

if __name__ == '__main__':
    executor = ProcessPoolExecutor()
    asyncio.get_event_loop().run_until_complete(start(executor))