异步与多处理相结合会出现什么样的问题（如果有）？

Question

几乎每个人都知道，当他们初次使用Python进行线程处理时，GIL使真正想并行执行处理的人痛苦不堪-至少给了机会。

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

我（尚）不确定我可能要面对什么样的负担。我知道目前对传入邮件设置了2MB的上限。从理论上讲，我们每秒可以得到数千个（尽管我不知道实际上是否已经看到过类似的东西）。处理消息所花费的时间并不是
很 重要，尽管显然更快会更好。

我正在研究Reactor模式，并使用该multiprocessing库（至少在测试中）开发了一个小示例，该示例似乎运行良好。但是，现在/不久我们将拥有可用的asyncio库，该库将为我处理事件循环。

通过组合使用asyncio，有什么可以咬我的multiprocessing吗？

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())

对于大多数情况，仅此功能就足够了。如果你发现自己需要其他结构的multiprocessing，如Queue，Event，Manager，等，有一个所谓的第三方库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()