不能直接用threading.Thread跑open_download_stream()，因为PyMongo 4.0+的GridFSBucket该方法返回协程对象，需await调度；在线程中调用会静默丢chunk、导致文件截断且无报错，官方要求必须配合asyncio使用。
为什么不能直接用 threading.Thread 跑多个 open_download_stream()
GridFS 的 open_download_stream() 返回的是异步/协程友好的流对象（PyMongo 4.0+ 默认基于 asyncio），在同步线程里直接调用会触发 RuntimeWarning: coroutine 'GridFSBucket.open_download_stream' was never awaited，且静默丢 chunk —— 表现为文件末尾截断、MD5 不匹配，但无报错。
根本原因：PyMongo 的 GridFSBucket 在同步模式下已弃用，官方明确要求使用 asyncio 配合 await；强行在线程中调用协程函数，Python 不会自动调度，底层 socket 读取直接跳过。
- 别写
threading.Thread(target=fs.open_download_stream, args=(file_id,))—— 这根本不会执行下载逻辑 - 也别用
concurrent.futures.ThreadPoolExecutor包裹open_download_stream()—— 协程对象无法被线程池“运行”，只会返回未 await 的挂起对象 - 真正能并发的，是
asyncio.create_task()+asyncio.gather()，不是线程
正确做法：用 asyncio + asyncio.Semaphore 控制并发数
GridFS 下载天然适合异步 I/O，瓶颈在 MongoDB 网络往返和 chunk 解包，而非 CPU。用 asyncio 不仅安全，还能压满连接池吞吐量。
关键控制点是限制同时打开的流数量（避免打爆 maxPoolSize 或触发 MongoDB 的 maxConns 限流）：
- 用
asyncio.Semaphore(3)限制最多 3 个并发下载流（推荐值：3–5，视 MongoDB 部署和网络延迟调整） - 每个任务必须
await完整流式读取，不能只await fs.open_download_stream()就结束 - 必须显式
stream.close()或用async with，否则流不释放，连接池耗尽后新请求卡住
示例片段：
import asyncio
from pymongo import MongoClient
from gridfs import GridFSBucket
<p>client = MongoClient("mongodb://localhost:27017")
db = client["mydb"]
fs = GridFSBucket(db)</p><p>async def download_one(file_id, filename):
async with semaphore: # 控制并发
try:
stream = await fs.open_download_stream(file_id)
with open(filename, "wb") as f:
while True:
chunk = await stream.read(65536) # 每次读 64KB
if not chunk:
break
f.write(chunk)
await stream.close() # 必须 close，否则连接泄漏
except Exception as e:
print(f"Download failed for {file_id}: {e}")</p><p>semaphore = asyncio.Semaphore(3)
tasks = [download_one(id1, "a.mp4"), download_one(id2, "b.pdf")]
await asyncio.gather(*tasks)</p>
多文件批量下载时，如何避免内存爆炸
常见错误是先 await fs.find({}) 拿到全部 GridOut 对象再并发，但 GridOut 构造时会预加载部分元数据并持有引用，大量文件会累积内存。
librosa 是一个用于音频和音乐分析的 Python 库。它提供了特征提取、频谱分析、节拍跟踪、起始检测和音频可视化的工具，基于 NumPy 和 SciPy 构建，用于科学音频计算。
更轻量的做法是只查 _id 和 filename，不构造完整对象：
- 用
fs.find({}, projection={"_id": 1, "filename": 1})，避免传输length、uploadDate等冗余字段 - 不要用
list(await fs.find().to_list(length=None))—— 这会一次性把所有文档拉进内存 - 改用游标迭代：
async for doc in fs.find({}): file_id = doc["_id"]，边取边 dispatch 任务 - 如果要限速，直接在游标循环里
await asyncio.sleep(0.01)，比堆满任务队列更可控
并发下载失败后，流和连接会不会卡住
会。PyMongo 的下载流一旦出错（如 MongoServerError: File not found 或网络中断），若没手动销毁，底层 socket 连接可能长期空闲占用，直到超时（默认 30s）。高并发场景下，几十个残留流就能吃光连接池。
必须做两件事：
- 每个下载任务外层加
try/except，捕获异常后立即if 'stream' in locals(): await stream.close() - 给整个
asyncio.gather()加 timeout：await asyncio.wait_for(asyncio.gather(*tasks), timeout=300)，防止单个慢请求拖垮全部 - 别依赖
__del__或 GC 回收流 —— 异步资源必须显式释放
最隐蔽的坑是：你以为只是“某个文件下载失败”，实际是后续所有请求都开始变慢 —— 那大概率是连接池被未关闭的流占满了。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
