中文
devexamples

Limit Async Task Fan-Out With Semaphore

Run many async coroutines while capping concurrency with a semaphore, avoiding the flood of pending tasks that unbounded gathering creates.

PythonResilience

Code

snippet
import asyncio
from typing import Awaitable, List, TypeVar

T = TypeVar("T")


async def gather_limited(
    tasks: List[Awaitable[T]], limit: int
) -> List[T]:
    if limit <= 0:
        raise ValueError("limit must be positive")
    semaphore = asyncio.Semaphore(limit)

    async def guarded(task: Awaitable[T]) -> T:
        async with semaphore:
            return await task

    return await asyncio.gather(*(guarded(t) for t in tasks))


async def fetch(i: int) -> str:
    await asyncio.sleep(0.01)
    return f"item-{i}"


async def main() -> None:
    results = await gather_limited([fetch(i) for i in range(20)], 5)
    print(len(results))

Related Examples