本文介绍如何使用 `asyncio.as_completed()` 动态管理可变长度任务队列，始终保持 n 个并发任务运行；当任务失败时立即重新入队并获得最高优先级，确保快速重试，避免批量提交导致的调度延迟。
在实际异步任务调度场景中（如网络请求、文件下载或批量 API 调用），我们常需满足三个关键约束：
- 固定并发上限（例如最多同时运行 8 个任务）；
- 失败任务需立即重试且享有最高优先级；
- 结果需按完成顺序实时产出（而非等待整批结束）。
此时，若一次性创建全部 1000 个 Task（如 asyncio.create_task() 批量调用），不仅内存开销大，更会导致失败任务被“埋没”——因为 asyncio.as_completed(task_list) 仅作用于已创建的 Task 对象集合，无法动态插入新任务或调整优先级。
✅ 正确解法是：只维护一个大小可控的活跃任务池（pool），通过循环“取—执行—回收—补充”实现流式调度。核心思路如下：
- 使用字典 task_pool: Dict[asyncio.Task, Coroutine] 映射任务与其原始协程，便于失败后精准重试；
- 每次循环优先填充至目标并发数（如 8）；
- 调用 asyncio.as_completed(task_pool.keys()) 获取首个完成任务；
- await 其结果，成功则 yield，失败则将对应协程重新加入待调度队列（推荐用 list.append() 实现 LIFO 高优先级）；
- 无论成败，均从池中移除该任务，并在下一轮循环中自动补入新任务。
以下是生产就绪的参考实现（含错误处理与资源清理）：
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），需与本层重试策略对齐，避免重复或冲突。
该模式已被广泛应用于异步爬虫、批量数据同步及微服务熔断降级等场景，兼顾性能、可控性与健壮性。
