Python異步之迭代器怎么使用

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

正文

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

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

1. 什么是異步迭代器

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

1.1. Iterators

迭代器是實現(xiàn)特定接口的 Python 對象。具體來說，返回迭代器實例的 iter() 方法和使迭代器步進一個循環(huán)并返回值的 next() 方法?？梢允褂脙?nèi)置函數(shù) next() 步進迭代器或使用 for 循環(huán)遍歷迭代器。許多 Python 對象是可迭代的，最值得注意的是列表等容器。

1.2. Asynchronous Iterators

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

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

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

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

可以使用 anext() 內(nèi)置函數(shù)步進異步迭代器，該函數(shù)返回執(zhí)行迭代器一步的可等待對象，例如一次調(diào)用 anext() 方法。

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

2. 什么是“async for”循環(huán)？

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

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

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

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

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

這不會并行執(zhí)行 for 循環(huán)。 asyncio 無法在一個 Python 線程中一次執(zhí)行多個協(xié)程。

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

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

這將構(gòu)建異步迭代器的返回值列表。

3. 如何使用異步迭代器

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

• 定義異步迭代器

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

因為異步迭代器是一個協(xié)程，并且每個迭代器返回一個在 asyncio 事件循環(huán)中調(diào)度和執(zhí)行的等待對象，所以我們可以在迭代器的主體內(nèi)執(zhí)行和等待等待對象。

...
# 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

• 創(chuàng)建異步迭代器

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

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

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

• 迭代一個異步迭代器

可以使用 anext() 內(nèi)置函數(shù)遍歷迭代器的一步，就像使用 next() 函數(shù)的經(jīng)典迭代器一樣。結(jié)果是等待的可等待對象。

...
# 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

這可以一步實現(xiàn)。

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

• 遍歷異步迭代器

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

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

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

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

4. 異步迭代器示例

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

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

# 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())

運行示例首先創(chuàng)建 main() 協(xié)程并將其用作 asyncio 程序的入口點。main() 協(xié)程運行并啟動 for 循環(huán)。

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

一旦迭代器的內(nèi)部計數(shù)器達到 10，就會引發(fā) StopAsyncIteration。這不會終止程序。相反，它由“async for”表達式預(yù)期和處理并中斷循環(huán)。

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

1
2
3
4

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