多处理进程中的 Python AsyncIO

Question

我正在尝试创建两个永远运行的进程，每个进程在它们内部运行一个异步循环。

我想了解逻辑是否正确。有没有更好的方法来做同样的事情？

import asyncio
import multiprocessing


async def my_async_func(topic):
    while True:
        await asyncio.sleep(5)
        print(topic)


def create_aio_loop(topic):
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(my_async_func(topic), loop=loop)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        loop.stop()


def main():
    topic_a = 'New to Asyncio'
    topic_b = 'New to Multiprocessing'
    process_a = multiprocessing.Process(target=create_aio_loop, args=(topic_a, ))
    process_b = multiprocessing.Process(target=create_aio_loop, args=(topic_b, ))
    processes = [process_a, process_b]

    try:
        for process in processes:
            process.start()
    except KeyboardInterrupt:
        for process in processes:
            process.terminate()
            process.join()


if __name__ == '__main__':
    main()

Answer 1

我正在尝试创建两个永远运行的进程，每个进程在它们内部运行一个异步循环。

我假设您知道为什么要在多个处理（虚拟）核心（多处理）上分派一些代码，并将其余代码并行化在同一个核心（异步）上。

那么我认为你做对了：你产生了两个进程，每个进程都有自己的异步循环。我能找到的唯一改进是使用loop.run_until_complete，它删除了一行代码:)：

import os
import asyncio
import multiprocessing

async def my_async_func(topic):
    while True:
        await asyncio.sleep(5)
        print(topic)


def create_aio_loop(topic):
    process_name = "[Process %s, topic %s]" % (os.getpid(), topic)
    print("%s Started " % process_name)

    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(my_async_func(topic))
    except KeyboardInterrupt:
        print("%s Loop interrupted" % process_name)
        loop.stop()

    print("%s terminating" % process_name)


if __name__ == '__main__':
    topic_a = 'New to Asyncio'
    topic_b = 'New to Multiprocessing'
    process_a = multiprocessing.Process(target=create_aio_loop, args=(topic_a, ))
    process_b = multiprocessing.Process(target=create_aio_loop, args=(topic_b, ))
    processes = [process_a, process_b]

    try:
        for process in processes:
            process.start()
    except KeyboardInterrupt:
        for process in processes:
            process.terminate()
            process.join()

哦，我还建议您使用包含进程 ID 的前缀显示所有消息，这对于多处理调试来说更容易。我介绍了一个带有开始/终止打印消息的示例。

运行这个会产生：

>python tmp_asyncio.py
[Process 11456, topic New to Multiprocessing] Started
[Process 18396, topic New to Asyncio] Started
New to Asyncio
New to Multiprocessing

(here I pressed Ctrl+C)

[Process 11456, topic New to Multiprocessing] Loop interrupted
[Process 11456, topic New to Multiprocessing] terminating
[Process 18396, topic New to Asyncio] Loop interrupted
[Process 18396, topic New to Asyncio] terminating
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "C:\Miniconda3\envs\baseenv\lib\multiprocessing\util.py", line 310, in _exit_function
    p.join()
  File "C:\Miniconda3\envs\baseenv\lib\multiprocessing\process.py", line 121, in join
    res = self._popen.wait(timeout)
  File "C:\Miniconda3\envs\baseenv\lib\multiprocessing\popen_spawn_win32.py", line 81, in wait
    res = _winapi.WaitForSingleObject(int(self._handle), msecs)
KeyboardInterrupt