本文介绍一种基于 asyncio 协程池与实时磁盘空间监控的动态任务调度方案，通过为每个 worker 独立检查剩余空间并按需拉取待下载文件夹，实现在磁盘空闲空间低于 200gb 时自动停止新任务启动，同时确保已在执行的任务完成。
本文介绍一种基于 asyncio 协程池与实时磁盘空间监控的动态任务调度方案，通过为每个 worker 独立检查剩余空间并按需拉取待下载文件夹，实现在磁盘空闲空间低于 200gb 时自动停止新任务启动，同时确保已在执行的任务完成。
要实现“下载过程中持续监控磁盘空间，并在剩余空间低于 200GB 时停止新增下载任务，但不中断正在运行的任务”，关键在于解耦任务调度与空间检查逻辑——不能依赖固定分块（如每 100 个一批），而应采用「生产者-消费者」式协程池模型：多个 worker 并发从共享任务队列中安全取任务，每次取前实时校验磁盘空间。
以下是一个简洁、健壮且线程/协程安全的实现方案：
import asyncio
from typing import List, Deque
from collections import deque
import aioshutil
# 全局常量：200 GB → 字节
DISK_LIMIT_BYTES = 200 * 10**9
async def download_folder(folder: str, session) -> None:
# 此处实现你的实际下载逻辑（如 aiohttp GET + 文件写入）
pass
async def worker(queue: Deque[str], session) -> None:
"""单个工作协程：循环检查空间、取任务、执行下载"""
while queue and len(queue) > 0:
# 每次执行前都检查磁盘空间
_, _, free = await aioshutil.disk_usage("/media/hdd")
if free None:
# 初始化双端队列（支持 O(1) 左侧弹出）
folder_queue = deque(folders_to_download)
async with aiohttp.ClientSession() as session:
# 启动 100 个并发 worker（可根据 CPU/IO 能力调整）
workers = [worker(folder_queue, session) for _ in range(100)]
await asyncio.gather(*workers, return_exceptions=True)
print("所有允许范围内的下载任务已完成。")
✅ 核心优势说明：
- ✅ 细粒度控制：每个 worker 在获取下一个任务前独立检查磁盘空间，响应更快，避免整批阻塞；
- ✅ 自然终止：当任意 worker 发现空间不足时，仅停止自身后续取任务；其余仍在执行的任务不受影响，保证已启动下载完成；
- ✅ 无锁安全：使用 deque + popleft() 在单线程事件循环中天然线程安全（无需 asyncio.Lock）；
- ✅ 可扩展性强：若需更复杂调度（如优先级、重试、限速），可无缝替换为 asyncio.Queue 并配合 get()/task_done() 使用。
⚠️ 注意事项：
- aioshutil.disk_usage 是异步调用，务必 await，不可用同步 shutil.disk_usage 替代；
- 若 folders_to_download 极大（百万级），建议改用 asyncio.Queue 配合 queue.put_nowait() 预填充，避免内存压力；
- 实际部署时，建议添加异常捕获与日志（如下载失败重试、空间检查超时），并考虑磁盘缓存导致的 free 延迟（可加 5–10GB 缓冲余量）；
- asyncio.gather(*workers) 默认会等待所有 worker 结束；若需提前感知“全部停止”，可封装为 asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED) + 状态标志。
该方案摒弃了静态分块的低效性，转而以资源可用性为驱动，真正实现了「空间够就持续下载，不够就优雅收尾」的自适应异步流控，是 I/O 密集型下载场景下的推荐实践。
