使用Asyncio并行化发电机

人气:591 发布:2022-10-16 标签: python multithreading generator python-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) + '
')

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

现在我想在Asyncio的帮助下并行化这项任务,但我似乎想不出如何实现。对我来说,主要问题是直接通过生成器将数据从生产者提供给消费者,同时让asyncio向io_task(x)发出多个并行请求。此外,async def@asyncio.coroutine这整件事让我感到困惑。

谁能向我演示如何使用此示例代码中的asyncio构建一个最小的工作示例?

(注意:只调用io_task(),缓冲结果,然后将结果写入文件是不行的。我需要一个可以处理超过主内存的大数据集的解决方案,这就是我到目前为止一直使用生成器的原因。然而,可以肯定的是,消费者的速度总是比所有生产者的速度加起来都快)

推荐答案

从Python3.6和asynchronous generators开始,只需进行少量更改即可使您的代码与Asyncio兼容。

io_task函数变为协程:

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

producer生成器变为异步生成器:

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

consumer函数变为协程并使用aiofiles、异步上下文管理和异步迭代:

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

并且主协程在事件循环中运行:

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

此外,您还可以考虑使用aiostream在生产者和消费者之间传递一些处理操作。

编辑:使用as_completed:

可以很容易地在生产者端并发运行不同的I/O任务
async def producer(xs):
    coros = [io_task(x) for x in xs]
    for future in asyncio.as_completed(coros):
        yield await future

763