如何在不阻塞事件循环的情况下迭代大型列表

Posted 2023-02-03

我有一个运行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

答案

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

据推测，列表非常大，这就是为什么您需要特殊措施来解锁事件循环。但是如果列表太大，强制每个循环迭代产生事件循环将slow it down considerably。当然，你可以通过添加一个计数器来缓解这种情况，只在i%10 == 0或i%100 == 0等时等待。但是你必须做出关于放弃控制的频率的任意决定（猜测）。如果你经常屈服，那么你的功能就会变慢。如果你很少屈服，那么你就会让事件循环无法响应。

如RafaëlDera所建议的那样，使用run_in_executor可以避免这种情况。 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)

另一答案

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

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