在Python中，是否有async相当于multiprocessing或concurrent.futures？

基本上，我正在寻找使用 python3 协程作为后端而不是线程或进程提供并行映射的东西。我相信执行高度并行 IO 工作时应该有更少的开销。

当然类似的东西已经存在了，无论是在标准库中还是在一些广泛使用的包中？

免责声明 PEP 0492 https://www.python.org/dev/peps/pep-0492/仅定义协程的语法和用法。它们需要一个事件循环来运行，这很可能是asyncio的事件循环 https://docs.python.org/3/library/asyncio-eventloop.html.

异步地图

我不知道有任何实施map基于协程。然而，实现基本的功能是微不足道的map功能使用asyncio.gather() https://docs.python.org/3/library/asyncio-task.html#asyncio.gather:

def async_map(coroutine_func, iterable):
    loop = asyncio.get_event_loop()
    future = asyncio.gather(*(coroutine_func(param) for param in iterable))
    return loop.run_until_complete(future)

这个实现非常简单。它为中的每个项目创建一个协程iterable，将它们加入单个协程并在事件循环上执行加入的协程。

提供的实施涵盖了部分案例。然而它有一个问题。对于长迭代，您可能希望限制并行运行的协程数量。我无法想出简单的实现，既高效又保持顺序，所以我将其作为练习留给读者。

表现

您声称：

我相信执行高度并行 IO 工作时应该有更少的开销。

它需要证明，所以这里有一个比较multiprocessing执行，gevent实施者a p https://stackoverflow.com/a/33007988/1377864以及我基于协程的实现。所有测试均在 Python 3.5 上进行。

实现使用multiprocessing:

from multiprocessing import Pool
import time


def async_map(f, iterable):
    with Pool(len(iterable)) as p:  # run one process per item to measure overhead only
        return p.map(f, iterable)

def func(val):
    time.sleep(1)
    return val * val

实现使用gevent:

import gevent
from gevent.pool import Group


def async_map(f, iterable):
    group = Group()
    return group.map(f, iterable)

def func(val):
    gevent.sleep(1)
    return val * val

实现使用asyncio:

import asyncio


def async_map(f, iterable):
    loop = asyncio.get_event_loop()
    future = asyncio.gather(*(f(param) for param in iterable))
    return loop.run_until_complete(future)

async def func(val):
    await asyncio.sleep(1)
    return val * val

测试程序正常timeit:

$ python3 -m timeit -s 'from perf.map_mp import async_map, func' -n 1 'async_map(func, list(range(10)))'

Results:

1. 可迭代的10 items:

   • multiprocessing- 1.05 秒
   • gevent - 1 sec
   • asyncio - 1 sec

2. 可迭代的100 items:

   • multiprocessing- 1.16 秒
   • gevent- 1.01 秒
   • asyncio- 1.01 秒

3. 可迭代的500 items:

   • multiprocessing- 2.31 秒
   • gevent- 1.02 秒
   • asyncio- 1.03 秒

4. 可迭代的5000 items:

   • multiprocessing - failed（生成 5k 个进程并不是一个好主意！）
   • gevent- 1.12 秒
   • asyncio- 1.22 秒

5. 可迭代的50000 items:

   • gevent- 2.2秒
   • asyncio- 3.25 秒

结论

当程序主要执行 I/O 而不是计算时，基于事件循环的并发运行速度更快。请记住，当 I/O 较少且涉及的计算较多时，差异会更小。

生成进程引入的开销明显大于基于事件循环的并发引入的开销。这意味着你的假设是正确的。

比较asyncio and gevent我们可以说asyncio开销增加 33-45%。这意味着创建 greenlet 比创建协程更便宜。

作为最终结论：gevent具有更好的性能，但是asyncio是标准库的一部分。性能差异（绝对数字）并不是非常显着。gevent是相当成熟的库，同时asyncio相对较新，但进步很快。