使用 asyncio 并行化生成器

我的应用程序从慢速 I/O 源读取数据，进行一些处理，然后将其写入本地文件。我已经用生成器实现了这个，如下所示：

import time

def io_task(x):
    print("requesting data for input %s" % x)
    time.sleep(1)   # this simulates a blocking I/O task
    return 2*x

def producer(xs):
    for x in xs:
        yield io_task(x)

def consumer(xs):
    with open('output.txt', 'w') as fp:
        for x in xs:
            print("writing %s" % x)
            fp.write(str(x) + '\n')

data = [1,2,3,4,5]
consumer(producer(data))

现在我想在 asyncio 的帮助下并行化这个任务，但我似乎不知道如何做。对我来说，主要问题是通过生成器直接将数据从生产者提供给消费者，同时让 asyncio 发出多个并行请求io_task(x)。还有，这整个async def vs. @asyncio.coroutine事情让我很困惑。

有人可以告诉我如何构建一个使用的最小工作示例吗asyncio从这个示例代码？

（注：这是not可以打电话给io_task()，缓冲结果然后将它们写入文件。我需要一个适用于可能超出主内存的大型数据集的解决方案，这就是我到目前为止一直使用生成器的原因。然而，可以安全地假设消费者总是比所有生产者加起来更快）

从 python 3.6 开始异步发电机 https://www.python.org/dev/peps/pep-0525/，只需进行很少的更改即可使您的代码与 asyncio 兼容。

The io_task函数变成协程：

async def io_task(x):
    await asyncio.sleep(1)
    return 2*x

The producer发电机变成异步发电机：

async def producer(xs):
    for x in xs:
        yield await io_task(x)

The consumer函数成为协程并使用aiofiles https://pypi.python.org/pypi/aiofiles、异步上下文管理和异步迭代：

async def consumer(xs):
    async with aiofiles.open('output.txt', 'w') as fp:
        async for x in xs:
            await fp.write(str(x) + '\n')

主协程在事件循环中运行：

data = [1,2,3,4,5]
main = consumer(producer(data))
loop = asyncio.get_event_loop()
loop.run_until_complete(main)
loop.close()

另外，您可以考虑使用音频流 https://github.com/vxgmichel/aiostream在生产者和消费者之间管道化一些处理操作。

编辑：不同的 I/O 任务可以通过使用轻松地在生产者端同时运行已完成 https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.as_completed:

async def producer(xs):
    coros = [io_task(x) for x in xs]
    for future in asyncio.as_completed(coros):
        yield await future