不能直接用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 限流):
立即学习“Python免费学习笔记(深入)”;
- 用
asyncio.Semaphore(3)限制最多 3 个并发下载流(推荐值:3–5,视 MongoDB 部署和网络延迟调整) - 每个任务必须
await完整流式读取,不能只await fs.open_download_stream()就结束 - 必须显式
stream.close()或用async with,否则流不释放,连接池耗尽后新请求卡住
示例片段:
librosa 是一个用于音频和音乐分析的 Python 库。它提供了特征提取、频谱分析、节拍跟踪、起始检测和音频可视化的工具,基于 NumPy 和 SciPy 构建,用于科学音频计算。
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)多文件批量下载时,如何避免内存爆炸
常见错误是先 await fs.find({}) 拿到全部 GridOut 对象再并发,但 GridOut 构造时会预加载部分元数据并持有引用,大量文件会累积内存。
更轻量的做法是只查 _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 回收流 —— 异步资源必须显式释放
最隐蔽的坑是:你以为只是“某个文件下载失败”,实际是后续所有请求都开始变慢 —— 那大概率是连接池被未关闭的流占满了。

















