如何在不阻塞事件循环的情况下迭代一个大列表答案

【问题标题】：How to iterate over a large list without blocking event loop如何在不阻塞事件循环的情况下迭代一个大列表
【发布时间】：2019-09-12 17:46:37
【问题描述】：

我有一个正在运行 asyncio 事件循环的 python 脚本，我想知道如何在不阻塞事件循环的情况下迭代一个大列表。从而保持循环运行。

我尝试使用 __aiter__ 和 __anext__ 创建一个自定义类，但没有成功，我还尝试创建一个 async function 产生结果但它仍然阻塞。

目前：

for index, item in enumerate(list_with_thousands_of_items):
    # do something

我尝试过的自定义类：

class Aiter:
    def __init__(self, iterable):
        self.iter_ = iter(iterable)

    async def __aiter__(self):
        return self

    async def __anext__(self):
        try:
            object = next(self.iter_)
        except StopIteration:
            raise StopAsyncIteration
        return object

但这总是会导致

TypeError: 'async for' received an object from __aiter__ that does not implement __anext__: coroutine

我制作的async function 可以工作但仍然阻塞事件循环是：

async def async_enumerate(iterable, start:int=0):
    for idx, i in enumerate(iterable, start):
        yield idx, i

【问题讨论】：

你对列表中的每一项都做了什么？
你考虑过类似 ProcessPoolExecutor 的东西吗？它可能非常适合您的问题。
@rdas 我将每个项目添加到一个字符串中
@RafaëlDera 很遗憾，我从未听说过 ProcessPoolExecutor
您的__aiter__ 应该是def，而不是async def。

标签： python python-3.x list asynchronous python-asyncio

【解决方案1】：

正如@deceze 指出的那样，您可以使用await asyncio.sleep(0) 将控制权显式传递给事件循环。不过，这种方法存在一些问题。

大概列表很大，这就是为什么您需要特殊措施来解除对事件循环的阻塞。但是如果列表太大，强制每个循环迭代屈服于事件循环将slow it down considerably。当然，您可以通过添加一个计数器并仅在 i%10 == 0 或 i%100 == 0 等时等待来缓解这种情况。但是您必须就放弃控制的频率做出任意决定（猜测）。如果你过于频繁地让步，你就会减慢你的功能。如果您的屈服太少，您就会使事件循环无响应。

这可以通过使用run_in_executor 来避免，正如 RafaëlDera 所建议的那样。 run_in_executor 接受阻塞函数并将其执行卸载到线程池。它立即返回一个可以在 asyncio 中为 awaited 的未来，其结果一旦可用，将是阻塞函数的返回值。（如果阻塞函数引发，异常将被传播。）这样的await 将暂停协程，直到函数返回或在其线程中引发，从而允许事件循环同时保持完整的功能。由于阻塞函数和事件循环在不同的线程中运行，因此该函数不需要做任何事情来允许事件工作运行——它们独立运行。即使 GIL 在这里也不是问题，因为 GIL 确保控制在线程之间传递。

使用run_in_executor，您的代码可能如下所示：

def process_the_list():
    for index, item in enumerate(list_with_thousands_of_items):
        # do something

loop = asyncio.get_event_loop()
await loop.run_in_executor(None, process_the_list)

【讨论】：

【解决方案2】：

asyncio 是合作多任务处理。合作部分来自这样一个事实，即您的函数必须让执行回到事件循环以允许其他事情运行。除非你 await 某事（或结束你的函数），否则你会占用事件循环。

您可以简单地await 一些noop 事件，可能最合适的是await asyncio.sleep(0)。这可确保您的任务尽快恢复，但也允许安排其他任务。

【讨论】：