如何在异步任务队列中动态维持固定并发数并优先重试失败任务

本文介绍如何使用 `asyncio.as_completed()` 动态管理可变长度任务队列，始终保持 n 个并发任务运行；当任务失败时立即重新入队并获得最高优先级，确保快速重试，避免批量提交导致的调度延迟。

在实际异步任务调度场景中（如网络请求、文件下载或批量 API 调用），我们常需满足三个关键约束：

• 固定并发上限（例如最多同时运行 8 个任务）；
• 失败任务需立即重试且享有最高优先级；
• 结果需按完成顺序实时产出（而非等待整批结束）。

此时，若一次性创建全部 1000 个 Task（如 asyncio.create_task() 批量调用），不仅内存开销大，更会导致失败任务被“埋没”——因为 asyncio.as_completed(task_list) 仅作用于已创建的 Task 对象集合，无法动态插入新任务或调整优先级。

✅ 正确解法是：只维护一个大小可控的活跃任务池（pool），通过循环“取—执行—回收—补充”实现流式调度。核心思路如下：

1. 使用字典 task_pool: Dict[asyncio.Task, Coroutine] 映射任务与其原始协程，便于失败后精准重试；
2. 每次循环优先填充至目标并发数（如 8）；
3. 调用 asyncio.as_completed(task_pool.keys()) 获取首个完成任务；
4. await 其结果，成功则 yield，失败则将对应协程重新加入待调度队列（推荐用 list.append() 实现 LIFO 高优先级）；
5. 无论成败，均从池中移除该任务，并在下一轮循环中自动补入新任务。

以下是生产就绪的参考实现（含错误处理与资源清理）：

import asyncio
from typing import List, Coroutine, Any, Iterator

async def run_with_priority_retry(
    task_coros: List[Coroutine],
    max_concurrent: int = 8,
) -> Iterator[Any]:
    """
    动态维持 max_concurrent 个并发任务，失败任务立即重试（LIFO 优先）。

    Args:
        task_coros: 初始任务协程列表（可修改）
        max_concurrent: 最大并发数

    Yields:
        成功执行的结果
    """
    # 使用 list 模拟优先队列：append → 高优；pop() → 取最高优
    pending = list(task_coros)
    task_pool = {}  # asyncio.Task -> Coroutine

    while pending or task_pool:
        # ✅ 补充任务至满额
        while pending and len(task_pool) < max_concurrent:
            coro = pending.pop()  # LIFO：最后加入的最先执行（高优）
            task = asyncio.create_task(coro)
            task_pool[task] = coro

        if not task_pool:
            break

        # ✅ 等待任意一个完成（as_completed 返回迭代器，next 即首个）
        done, _ = await asyncio.wait(
            task_pool.keys(), 
            return_when=asyncio.FIRST_COMPLETED
        )
        completed_task = done.pop()

        # ✅ 提取原始协程并清理池
        coro = task_pool.pop(completed_task)

        try:
            result = await completed_task
            yield result
        except Exception as e:
            print(f"Task failed with {type(e).__name__}: {e} — re-queued with high priority")
            pending.append(coro)  # 失败任务插到末尾，下次 pop 优先执行

        # ✅ 清理已完成 task 引用（防止内存泄漏）
        completed_task.cancel()
        try:
            await completed_task
        except (asyncio.CancelledError, Exception):
            pass

? 关键注意事项：

• ❗切勿用 asyncio.as_completed(list) 直接传入动态变化的列表——它接收的是快照，不支持运行时增删；应始终基于稳定集合（如 dict.keys()）调用。
• ⚠️ 使用 pending.pop() 而非 pending.pop(0) 实现 O(1) 高优入队/出队；若需严格 FIFO 重试，可改用 collections.deque 并调用 appendleft() + pop()。
• ? 务必显式 cancel() 并 await 已完成的 Task，避免悬空对象累积（尤其在长期运行服务中）。
• ? 若任务本身含重试逻辑（如 tenacity），需与本层重试策略对齐，避免重复或冲突。

该模式已被广泛应用于异步爬虫、批量数据同步及微服务熔断降级等场景，兼顾性能、可控性与健壮性。