中文
devexamples

Asyncio TaskGroup Timeout Gather

Fan out concurrent work with asyncio TaskGroup, wrapping the group in wait_for so one timeout cancels every straggler cleanly.

PythonResilience

Code

snippet
import asyncio


async def fetch_shard(shard: int) -> list[int]:
    await asyncio.sleep(0.2 * shard)  # 各分片耗时不一
    return [shard * 10, shard * 10 + 1]


async def fetch_slow_shard(shard: int) -> list[int]:
    await asyncio.sleep(10)
    return []


async def gather_with_timeout(coros, timeout: float) -> list[int]:
    async def runner():
        results: list[int] = []
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(c) for c in coros]
        results = [item for task in tasks for item in task.result()]
        return results

    try:
        return await asyncio.wait_for(runner(), timeout=timeout)
    except TimeoutError:
        print('deadline hit, all pending tasks cancelled')
        return []


async def main() -> None:
    fast = await gather_with_timeout(
        [fetch_shard(i) for i in range(4)], timeout=2.0
    )
    print('fast path collected:', fast)

    interrupted = await gather_with_timeout(
        [fetch_shard(1), fetch_slow_shard(9)], timeout=1.0
    )
    print('timeout path result:', interrupted)


asyncio.run(main())

Related Examples