因为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" 任务 + 定期扫描补偿逻辑。
- 在
claimed_at字段上建 TTL 索引：collection.create_index("claimed_at", expireAfterSeconds=300)，5 分钟没完成就自动删掉（或设为"timeout"） - 但直接删可能丢数据，更稳妥的是用 background job 定期扫描：
status == "running"且claimed_at 的任务，重置为 <code>"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。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
