Python異步之迭代器怎么使用

本文講解"Python異步之迭代器如何使用"，希望能夠解決相關問題。

正文

迭代是 Python 中的基本操作。我們可以迭代列表、字符串和所有其他結構。

Asyncio 允許我們開發異步迭代器。我們可以通過定義一個實現 aiter() 和 anext() 方法的對象來在 asyncio 程序中創建和使用異步迭代器。

1. 什么是異步迭代器

異步迭代器是一個實現了 aiter() 和 anext() 方法的對象。在我們仔細研究異步迭代器之前，讓我們回顧一下經典迭代器。

1.1. Iterators

迭代器是實現特定接口的 Python 對象。具體來說，返回迭代器實例的 iter() 方法和使迭代器步進一個循環并返回值的 next() 方法。可以使用內置函數 next() 步進迭代器或使用 for 循環遍歷迭代器。許多 Python 對象是可迭代的，最值得注意的是列表等容器。

1.2. Asynchronous Iterators

異步迭代器是實現特定接口的 Python 對象。異步迭代器必須實現 aiter() 和 anext() 方法。

• aiter() 方法必須返回迭代器的一個實例。

• anext() 方法必須返回一個步進迭代器的可等待對象。

異步迭代器只能在 asyncio 程序中步進或遍歷，例如在協程中。

可以使用 anext() 內置函數步進異步迭代器，該函數返回執行迭代器一步的可等待對象，例如一次調用 anext() 方法。

可以使用“async for”表達式遍歷異步迭代器，該表達式將在每次迭代時自動調用 anext() 并等待返回的 awaitable 以檢索返回值。

2. 什么是“async for”循環？

async for 表達式用于遍歷異步迭代器。它是一個異步的 for 循環語句。異步迭代器是產生可等待對象的迭代器。您可能還記得 awaitable 是可以等待的對象，例如協程或任務。

異步生成器將自動實現異步迭代器方法，允許它像異步迭代器一樣被迭代。await for 表達式允許調用者遍歷 awaitable 的異步迭代器并從每個迭代器中檢索結果。

這與遍歷集合或等待對象列表（例如協程對象）不同，相反，必須使用預期的異步迭代器方法提供返回的等待對象。在內部，async for 循環將根據需要自動解析或等待每個可等待的調度協程。

因為它是一個 for 循環，所以它假定（盡管不要求）每個被遍歷的等待對象都會產生一個返回值。async for 循環必須在協程內使用，因為它在內部會使用只能在協程內使用的 await 表達式。async for 表達式可用于在協程中遍歷異步迭代器。

...
# traverse an asynchronous iterator
async for item in async_iterator:
	print(item)

這不會并行執行 for 循環。 asyncio 無法在一個 Python 線程中一次執行多個協程。

相反，這是一個異步 for 循環。不同的是，執行 for 循環的協程會暫停并在內部等待每個 awaitable。在幕后，這可能需要安排和等待協程，或者等待任務。我們也可以在列表理解中使用 async for 表達式。

...
# build a list of results
results = [item async for item async_iterator]

這將構建異步迭代器的返回值列表。

3. 如何使用異步迭代器

在本節中，我們將仔細研究如何在 asyncio 程序中定義、創建、步進和遍歷異步迭代器。讓我們從如何定義異步迭代器開始。

• 定義異步迭代器

我們可以通過定義一個實現了 aiter() 和 anext() 方法的類來定義一個異步迭代器。這些方法通常在 Python 對象上定義。重要的是，因為 anext() 函數必須返回一個可等待對象，所以它必須使用“async def”表達式定義。迭代完成后，anext() 方法必須引發 StopAsyncIteration 異常。

# define an asynchronous iterator
class AsyncIterator():
    # constructor, define some state
    def __init__(self):
        self.counter = 0
    # create an instance of the iterator
    def __aiter__(self):
        return self
    # return the next awaitable
    async def __anext__(self):
        # check for no further items
        if self.counter >= 10:
            raise StopAsyncIteration
        # increment the counter
        self.counter += 1
        # return the counter value
        return self.counter

因為異步迭代器是一個協程，并且每個迭代器返回一個在 asyncio 事件循環中調度和執行的等待對象，所以我們可以在迭代器的主體內執行和等待等待對象。

...
# return the next awaitable
async def __anext__(self):
    # check for no further items
    if self.counter >= 10:
        raise StopAsyncIteration
    # increment the counter
    self.counter += 1
    # simulate work
    await asyncio.sleep(1)
    # return the counter value
    return self.counter

• 創建異步迭代器

要使用異步迭代器，我們必須創建迭代器。這涉及正常創建 Python 對象。

...
# create the iterator
it = AsyncIterator()

這將返回一個“異步迭代器”，它是“異步迭代器”的一個實例。

• 迭代一個異步迭代器

可以使用 anext() 內置函數遍歷迭代器的一步，就像使用 next() 函數的經典迭代器一樣。結果是等待的可等待對象。

...
# get an awaitable for one step of the iterator
awaitable = anext(it)
# execute the one step of the iterator and get the result
result = await awaitable

這可以一步實現。

...
# step the async iterator
result = await anext(it)

• 遍歷異步迭代器

異步迭代器也可以使用“async for”表達式在循環中遍歷，該表達式將自動等待循環的每次迭代。

...
# traverse an asynchronous iterator
async for result in AsyncIterator():
	print(result)

我們還可以使用帶有“async for”表達式的異步列表理解來收集迭代器的結果。

...
# async list comprehension with async iterator
results = [item async for item in AsyncIterator()]

4. 異步迭代器示例

我們可以探索如何使用“async for”表達式遍歷異步迭代器。在此示例中，我們將更新之前的示例，以使用“async for”循環遍歷迭代器直至完成。

此循環將自動等待從迭代器返回的每個可等待對象，檢索返回值，并使其在循環體內可用，以便在這種情況下可以報告它。這可能是異步迭代器最常見的使用模式。

# SuperFastPython.com
# example of an asynchronous iterator with async for loop
import asyncio
# define an asynchronous iterator
class AsyncIterator():
    # constructor, define some state
    def __init__(self):
        self.counter = 0
    # create an instance of the iterator
    def __aiter__(self):
        return self
    # return the next awaitable
    async def __anext__(self):
        # check for no further items
        if self.counter >= 10:
            raise StopAsyncIteration
        # increment the counter
        self.counter += 1
        # simulate work
        await asyncio.sleep(1)
        # return the counter value
        return self.counter
# main coroutine
async def main():
    # loop over async iterator with async for loop
    async for item in AsyncIterator():
        print(item)
# execute the asyncio program
asyncio.run(main())

運行示例首先創建 main() 協程并將其用作 asyncio 程序的入口點。main() 協程運行并啟動 for 循環。

異步迭代器的一個實例被創建，循環使用 anext() 函數自動單步執行它以返回一個可等待對象。然后循環等待可等待對象并檢索一個值，該值可用于報告它的循環體。然后重復這個過程，掛起 main() 協程，執行迭代器和掛起的一個步驟，然后恢復 main() 協程，直到迭代器耗盡。

一旦迭代器的內部計數器達到 10，就會引發 StopAsyncIteration。這不會終止程序。相反，它由“async for”表達式預期和處理并中斷循環。

這突出顯示了如何使用 async for 表達式遍歷異步迭代器。

1
2
3
4

關于 "Python異步之迭代器如何使用" 就介紹到此。希望多多支持碩編程。