如何为生成器函数正确使用 asyncio？

Question

我一次读取数千个文件，对于每个文件，我需要在从每个文件中产生行之前对其执行操作。为了提高性能，我想我可以使用 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()

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

Answer 1

我觉得我在这里遗漏了一些非常明显的东西，它使我的异步尝试同步。

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

第二个也是更基本的问题是，与可以并行化同步代码的线程不同，asyncio 要求一切从头开始都是异步的。将async 添加到像open_files 这样的函数以使其异步是不够的。您需要检查代码并将任何阻塞调用（例如对 IO 的调用）替换为等效的异步原语。例如，连接网络端口应该使用open_connection，等等。如果您的 async 函数不等待任何内容，就像 open_files 的情况一样，它将像常规函数一样执行，并且您不会获得 asyncio 的任何好处。

由于您在常规文件上使用 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()