如何正确地将asyncio用于生成器函数？

如何解决如何正确地将asyncio用于生成器函数？

我要一次读取数千个文件，对于每个文件，我需要先执行操作，然后再从每个文件中产生行。为了提高性能，我想我可以使用asyncio来在等待读取新文件的同时对文件（并产生行）执行操作。

但是从打印语句中，我可以看到所有文件都已打开并收集，然后对每个文件进行迭代（与不使用asyncio的情况相同）。

我感觉这里缺少明显的东西，这使得我的异步尝试变得同步了。

import asyncio

async def open_files(file):
    with open(file) as file:
        # do stuff
        print('opening files')
        return x

async def async_generator():
    file_outputs = await asyncio.gather(*[open_files(file) for file in files])

    for file_output in file_ouputs:
        print('using open file')
        for row in file_output:
            # Do stuff to row
            yield row

async def main():
    async for yield_value in async_generator():
        pass

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

输出：

opening files
opening files
.
.
.
using open file
using open file

---

编辑

使用@ user4815162342提供的代码，我注意到，尽管速度快了3倍，但生成器产生的行集与没有并发的情况下稍有不同。我还不确定这是否是因为每个文件都错过了一些收益，还是因为文件被重新排序了。因此，我对user4815162342的代码进行了以下更改，并在pool.submit()

中输入了锁定

我应该在初次询问时提到每个文件中行的顺序以及文件本身的顺序。

import concurrent.futures

def open_files(file):
    with open(file) as file:
        # do stuff
        print('opening files')
        return x

def generator():
    m = multiprocessing.Manager()
    lock = m.Lock()
    pool = concurrent.futures.ThreadPoolExecutor()
    file_output_futures = [pool.submit(open_files,file,lock) for file in files]
    for fut in concurrent.futures.as_completed(file_output_futures):
        file_output = fut.result()
        print('using open file')
        for row in file_output:
            # Do stuff to row
            yield row

def main():
    for yield_value in generator():
        pass

if __name__ == '__main__':
    main()

这样我非并发和并行方式产生相同的值每一次，但是我刚刚失去了使用并发获得的所有速度。

解决方法

我感觉这里缺少明显的东西，这使得我的异步尝试变得同步了。

您的代码有两个问题。第一个是asyncio.gather()的设计是等待所有 all 期货并行完成，然后才返回其结果。因此，您在生成器中执行的处理不会像您的意图那样散布在open_files中的IO中，而是仅在返回所有对open_files的调用之后才开始。要在完成异步调用后对其进行处理，您应该使用asyncio.as_completed之类的东西。

第二个也是更基本的问题是，与可以并行化同步代码的线程不同，asyncio要求从头开始一切都是异步的。仅将async添加到类似open_files的函数中以使其异步是不够的。您需要遍历代码，并用等效的异步原语替换所有阻塞调用，例如对IO的调用。例如，连接网络端口应使用open_connection，依此类推。如果您的异步函数没有等待任何东西（例如open_files的情况），它将像常规函数一样执行，您将不会获得异步的任何好处。

由于您在常规文件上使用IO，并且操作系统不会为常规文件提供可移植的异步接口，因此您不太可能从asyncio中获利。有一些像aiofiles这样的库在内部使用线程，但它们使您的代码慢于加快代码的速度，因为它们漂亮的异步API涉及很多内部线程同步。为了加快代码速度，您可以使用经典的线程池，Python通过concurrent.futures模块公开了该线程池。例如（未试用）：

import concurrent.futures

def open_files(file):
    with open(file) as file:
        # do stuff
        print('opening files')
        return x

def generator():
    pool = concurrent.futures.ThreadPoolExecutor()
    file_output_futures = [pool.submit(open_files,file) for file in files]
    for fut in file_output_futures:
        file_output = fut.result()
        print('using open file')
        for row in file_output:
            # Do stuff to row
            yield row

def main():
    for yield_value in generator():
        pass

if __name__ == '__main__':
    main()