異步迭代，如何在等待任務完成的同時進行下一步迭代？

我正在嘗試撰寫一個迭代器，它在等待 IO 系結任務時繼續迭代的下一步。大致演示我在代碼中嘗試做的事情

for i in iterable:
    await io_bound_task()   # move on to next step in iteration
    # do more stuff when task is complete

我最初嘗試使用一個簡單的for回圈運行，sleep模擬一個 IO 系結任務

import asyncio
import random


async def main() -> None:
    for i in range(3):
        print(f"starting task {i}")
        result = await io_bound_task(i)
        print(f"finished task {result}")


async def io_bound_task(i: int) -> int:
    await asyncio.sleep(random.random())
    return i


asyncio.run(main())

這里代碼同步運行并輸出

starting task 0
finished task 0
starting task 1
finished task 1
starting task 2
finished task 2

我認為這是因為for回圈被阻塞了。所以我認為異步for回圈是繼續的方式？所以我嘗試使用異步迭代器

from __future__ import annotations

import asyncio
import random


class AsyncIterator:
    def __init__(self, max_value: int) -> None:
        self.max_value = max_value
        self.count = 0

    def __aiter__(self) -> AsyncIterator:
        return self

    async def __anext__(self) -> int:
        if self.count == self.max_value:
            raise StopAsyncIteration
        self.count  = 1
        return self.count


async def main() -> None:
    async for i in AsyncIterator(3):
        print(f"starting task {i}")
        result = await io_bound_task(i)
        print(f"finished task {result}")


async def io_bound_task(i: int) -> int:
    await asyncio.sleep(random.random())
    return i


asyncio.run(main())

但這似乎也同步運行并導致輸出

starting task 1
finished task 1
starting task 2
finished task 2
starting task 3
finished task 3

每次。所以我認為異步迭代器沒有做我認為它會做的事情？在這一點上，我被困住了。我對異步迭代器的理解有問題嗎？有人可以給我一些關于如何實作我想要做的事情的指示嗎？

I'm new to working with async, so apologies if I'm doing something stupid. Any help is appreciated. Thanks.

I'm on python 3.8.10 if that is a relevant detail.

uj5u.com熱心網友回復：

您正在尋找的東西稱為任務，可以使用該asyncio.create_task函式創建。您嘗試的所有方法都涉及等待 coroutine io_bound_task(i)，并且 await 的意思是“等待此完成后再繼續”。如果您將協程包裝在一個任務中，那么它將在后臺運行，而不是您必須等待它完成才能繼續。

這是使用任務的代碼版本：

import asyncio
import random


async def main() -> None:
    tasks = []
    for i in range(3):
        print(f"starting task {i}")
        tasks.append(asyncio.create_task(io_bound_task(i)))

    for task in tasks:
        result = await task
        print(f"finished task {result}")


async def io_bound_task(i: int) -> int:
    await asyncio.sleep(random.random())
    return i


asyncio.run(main())

輸出：

starting task 0
starting task 1
starting task 2
finished task 0
finished task 1
finished task 2

您還可以使用asyncio.gather（如果在繼續之前需要所有結果）或asyncio.wait等待多個任務，而不是回圈。例如，如果任務 2 在任務 0 之前完成并且您不想等待任務 0，則可以執行以下操作：

async def main() -> None:
    pending = []
    for i in range(3):
        print(f"starting task {i}")
        pending.append(asyncio.create_task(io_bound_task(i)))

    while pending:
        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            result = await task
            print(f"finished task {result}")

標籤：python asynchronous python-asyncio

上一篇：給定API端點串列及其優先級，回傳具有高優先級的API回應

下一篇：無法在DjangoAsyncconusmers中獲取模型資料。在頻道和websockets中作業