一个轻量异步任务调度器需支持并发控制、错误隔离、动态添加、结果统一收集与任务取消;通过 running 计数器限流,队列暂存待执行任务,try/catch 捕获结果,isCancelled 标志跳过未执行任务。

实现一个通用的异步任务调度器,核心在于控制并发、隔离错误、支持动态添加,并能统一收集结果。不需要依赖 Celery 或 APScheduler 这类重型框架,用原生 async/await 就能搭出轻量、可复用的调度器。
控制最大并发数
关键不是“一口气全发”,而是限制同时运行的任务数量。比如设为 3,那就最多有 3 个 Promise 在 pending 状态,其余排队等待。
- 维护一个计数器 running,记录当前正在执行的任务数
- 每次准备执行新任务前,检查 running < concurrency;不满足就先挂起(用
Promise或asyncio.Event等待) - 任务完成(无论成功或失败)后,running 减 1,并立即触发下一个待执行任务
任务队列与安全添加
任务应以函数形式传入,该函数返回 Promise(或协程),而不是直接执行。这样调度器才能按需调用。
- 用数组或队列(如
asyncio.Queue或普通数组)暂存待执行任务 - addTask(task) 方法只做入队,不立即执行;内部自动触发调度逻辑
- 添加时做基础校验:是否为函数、是否返回可 await 对象,避免静默失败
结果收集与错误隔离
一个任务失败不能让整个调度器崩掉,也不能阻塞后续任务。所有结果(包括异常)都应被捕获并归档。
- 每个任务用
try/catch或.catch()包裹,确保异常不冒泡 - 将结果(含
{ success: true, value }或{ success: false, error })推入统一数组 results - 提供 waitAll() 方法,返回一个 Promise,等全部任务(含排队中)结束才 resolve
支持取消未执行任务
用户可能中途放弃,这时已开始的任务照常运行(不强制中断),但排队中的任务应跳过。
- 引入布尔标志 isCancelled,
addTask前检查它 - 在调度循环中,每次取任务前也判断是否已取消,是则跳过该任务
- 可暴露 cancel() 方法,置位标志并清空队列(已运行的不受影响)

















