将 asyncio 与多处理结合起来会出现什么样的问题（如果有的话）？

Question

正如几乎每个人在第一次看到 Python 中的线程时都知道的那样，GIL 让真正想要并行处理的人的生活变得悲惨——或者至少给它一个机会。

我目前正在考虑实施类似 Reactor 模式的东西。实际上，我想在类似线程的情况下侦听传入的套接字连接，当有人尝试连接时，接受该连接并将其传递给类似线程的另一个线程进行处理。

我（还）不确定我可能面临什么样的负载。我知道目前对传入消息设置了 2MB 的上限。从理论上讲，我们每秒可以获得数千次（尽管我不知道实际上我们是否见过这样的事情）。处理一条消息所花费的时间并不是 特别 重要，但显然越快越好。

我正在研究 Reactor 模式，并使用 multiprocessing 库开发了一个小示例（至少在测试中）似乎工作得很好。但是，现在/很快我们就会有可用的 asyncio 库，它将为我处理事件循环。

结合 asyncio 和 multiprocessing 有什么可以咬我的吗？

原文由 Wayne Werner 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您应该能够安全地组合 asyncio 和 multiprocessing 而不会遇到太多麻烦，尽管您不应该直接使用 multiprocessing 。 asyncio （以及任何其他基于事件循环的异步框架）的主要错误是阻塞事件循环。如果您尝试直接使用 multiprocessing ，任何时候您阻塞等待子进程，您都会阻塞事件循环。显然，这是不好的。

避免这种情况的最简单方法是使用 BaseEventLoop.run_in_executor concurrent.futures.ProcessPoolExecutor 执行函数。 ProcessPoolExecutor 是使用 multiprocessing.Process 实现的进程池，但是 asyncio 内置支持在其中执行函数而不阻塞事件循环。这是一个简单的例子：

 import time
import asyncio
from concurrent.futures import ProcessPoolExecutor

def blocking_func(x):
   time.sleep(x) # Pretend this is expensive calculations
   return x * 5

@asyncio.coroutine
def main():
    #pool = multiprocessing.Pool()
    #out = pool.apply(blocking_func, args=(10,)) # This blocks the event loop.
    executor = ProcessPoolExecutor()
    out = yield from loop.run_in_executor(executor, blocking_func, 10)  # This does not
    print(out)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

对于大多数情况，仅此功能就足够了。 If you find yourself needing other constructs from multiprocessing , like Queue , Event , Manager , etc., there is a third-party名为 aioprocessing 库（完全公开：我写了它），它提供了 asyncio 所有 multiprocessing 数据结构的兼容版本。这是一个演示的示例：

 import time
import asyncio
import aioprocessing
import multiprocessing

def func(queue, event, lock, items):
    with lock:
        event.set()
        for item in items:
            time.sleep(3)
            queue.put(item+5)
    queue.close()

@asyncio.coroutine
def example(queue, event, lock):
    l = [1,2,3,4,5]
    p = aioprocessing.AioProcess(target=func, args=(queue, event, lock, l))
    p.start()
    while True:
        result = yield from queue.coro_get()
        if result is None:
            break
        print("Got result {}".format(result))
    yield from p.coro_join()

@asyncio.coroutine
def example2(queue, event, lock):
    yield from event.coro_wait()
    with (yield from lock):
        yield from queue.coro_put(78)
        yield from queue.coro_put(None) # Shut down the worker

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue = aioprocessing.AioQueue()
    lock = aioprocessing.AioLock()
    event = aioprocessing.AioEvent()
    tasks = [
        asyncio.async(example(queue, event, lock)),
        asyncio.async(example2(queue, event, lock)),
    ]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

原文由 dano 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

是的，有很多位可能（或可能不会）咬你。

• 当你运行类似 asyncio 的东西时，它期望在一个线程或进程上运行。这（本身）不适用于并行处理。您必须以某种方式分配工作，同时将 IO 操作（特别是套接字上的操作）留在单个线程/进程中。
• 虽然将各个连接移交给不同的处理程序进程的想法很好，但很难实现。第一个障碍是您需要一种方法来拉出连接 asyncio 而不关闭它。下一个障碍是您不能简单地将文件描述符发送到不同的进程，除非您使用来自 C 扩展的特定于平台（可能是 Linux）的代码。
• 请注意，已知 multiprocessing 模块会创建多个线程进行通信。大多数情况下，当您使用通信结构（例如 Queue s）时，会生成一个线程。不幸的是，这些线程并不是完全不可见的。例如，它们可能无法彻底拆除（当您打算终止程序时），但根据它们的数量，资源使用量可能会自行引起注意。

如果您真的打算处理各个进程中的各个连接，我建议检查不同的方法。例如，您可以将套接字置于侦听模式，然后同时并行地接受来自多个工作进程的连接。一旦工作人员处理完请求，它就可以接受下一个连接，因此与为每个连接分叉一个进程相比，您仍然使用更少的资源。例如，Spamassassin 和 Apache (mpm prefork) 可以使用这个 worker 模型。根据您的用例，它最终可能会更容易、更健壮。具体来说，您可以让您的工作人员在处理配置数量的请求后死亡，并由主进程重新生成，从而消除内存泄漏的大部分负面影响。

原文由 Helmut Grohne 发布，翻译遵循 CC BY-SA 3.0 许可协议