本文介绍了使用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
这篇关于使用Asyncio并行化发电机的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！
本站部分内容来源互联网,如果有图片或者内容侵犯您的权益请联系我们删除！