因为Celery依赖RabbitMQ或Redis等消息中间件,而实际环境仅有MongoDB,且任务量小、无需高吞吐与复杂路由,故选用MongoDB构建轻量级任务队列;其原子操作find_one_and_update配合TTL索引和复合索引,可实现“能跑、可查、不丢任务”的分布式调度。

为什么不用 celery 而要自己用 MongoDB 做任务队列?
因为 celery 依赖消息中间件(如 RabbitMQ、Redis),而你手头只有 MongoDB,或者任务量小、不需要高吞吐和复杂路由,只是想快速落地一个“能跑、可查、不丢任务”的轻量队列。MongoDB 的原子更新($set + $inc)和 TTL 索引足够支撑简单分布式任务调度。
find_one_and_update 是核心:如何安全地取走一个待执行任务?
关键不是“查出来再改”,而是“查+改”一步完成,避免多个 worker 同时取到同一个任务。必须用 find_one_and_update,且带上严格筛选条件和原子更新操作:
- 状态必须是
"pending",且最好加worker_id为空(或 null)的条件,防止误抢 - 更新动作要同时设置
status: "running"和claimed_at: datetime.utcnow() - 必须指定
return_document=ReturnDocument.AFTER,确保拿到的是已更新后的文档 - 加上
sort={"priority": -1, "created_at": 1}可实现优先级+先进先出
示例片段:
task = collection.find_one_and_update(
{"status": "pending", "worker_id": None},
{"$set": {"status": "running", "worker_id": "w-01", "claimed_at": datetime.utcnow()}},
return_document=ReturnDocument.AFTER,
sort=[("priority", -1), ("created_at", 1)]
)怎么防止任务卡死或 worker 崩溃后任务丢失?
靠两层保护:TTL 索引自动清理超时的 "running" 任务 + 定期扫描补偿逻辑。
微软官方的 Python 扩展,是 VS Code 安装量最高的扩展(209M+)。集成 IntelliSense(通过 Pylance)、调试(通过 Python Debugger)、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。
立即学习“Python免费学习笔记(深入)”;
- 在
claimed_at字段上建 TTL 索引:collection.create_index("claimed_at", expireAfterSeconds=300),5 分钟没完成就自动删掉(或设为"timeout") - 但直接删可能丢数据,更稳妥的是用 background job 定期扫描:
status == "running"且claimed_at < now() - 300的任务,重置为"pending" - worker 启动时最好也检查一遍自己上次留下的
worker_id任务,避免崩溃残留
并发写入冲突和性能瓶颈在哪?
瓶颈不在 MongoDB 本身,而在任务文档结构设计和索引覆盖度。
- 所有查询条件字段(
status、worker_id、priority、created_at)必须合建复合索引,否则find_one_and_update会全表扫描 - 避免在任务文档里存大字段(比如原始日志、二进制 payload),拆到 GridFS 或外部存储,只留 ID 引用
- 高频场景下,单集合竞争会明显——可按业务域分片(如
tasks_user/tasks_order),或用collection_name.format(shard_id)动态选集合
真正容易被忽略的,是 claimed_at 字段的时钟一致性:不同 worker 机器时间差超过 TTL 就会导致误判。别省事,强制所有节点跑 chrony 或同步 NTP。








