MongoDB中使用upsert=True插入无重复设备任务时如何避免竞态条件？

你的当前实现确实存在竞态条件，原因如下：
MongoDB的update_one(..., upsert=True)操作本身是原子的，但这个原子性是指针对查询条件匹配到的文档，要么更新要么插入的操作是原子的。但你的查询条件是{"devices": {'$elemMatch': {'$in': task['devices']}}}——当两个并发请求提交的任务设备列表有重叠，且初始集合中没有任何匹配的文档时，两个请求的查询都会返回空结果，进而都会触发upsert操作，最终导致两个冲突的任务被插入。

由于无法使用事务，我们可以通过设备占用锁集合的方式来实现原子性的设备抢占，具体步骤如下：

1. 创建设备锁集合 #

首先创建一个单独的集合（比如device_locks），用于跟踪每个设备的占用状态。每个文档的结构为：

{
    "_id": <设备ID>,  # 用设备ID作为主键，天然唯一
    "task_id": <任务ID>  # 占用该设备的任务ID
}

2. 原子性抢占所有所需设备 #

在插入新任务前，先尝试为所有需要的设备创建锁文档：

from bson.objectid import ObjectId
from pymongo.errors import DuplicateKeyError
from datetime import datetime, timedelta

task = {'devices': [1, 2, 3], 'name': 'my_new_task'}
task_id = ObjectId()  # 提前生成任务ID
lock_collection = db.device_locks

acquired_locks = []
success = True

try:
    for device_id in task['devices']:
        # 尝试插入锁文档，仅当设备未被占用时成功
        result = lock_collection.update_one(
            {'_id': device_id},
            {'$setOnInsert': {
                'task_id': task_id,
                'expireAt': datetime.utcnow() + timedelta(hours=24)  # 添加过期时间
            }},
            upsert=True
        )
        if result.upserted_id:
            acquired_locks.append(device_id)
        else:
            # 该设备已被占用，抢占失败
            success = False
            break

    if success:
        # 所有设备抢占成功，插入任务文档
        task['_id'] = task_id
        collection.insert_one(task)
        print('任务插入成功')
    else:
        print('任务插入失败，存在占用的设备')

finally:
    if not success and acquired_locks:
        # 抢占失败，释放已获取的设备锁
        lock_collection.delete_many({'_id': {'$in': acquired_locks}})

3. 任务完成后释放设备锁 #

当任务执行完成或取消时，记得从device_locks集合中删除对应的设备锁文档，以便其他任务可以使用这些设备：

lock_collection.delete_many({'task_id': task_id})

补充优化 #

为避免任务异常终止导致设备锁无法释放，可给锁文档添加TTL索引，让过期的锁自动失效：

# 为lock_collection创建TTL索引，24小时后自动删除过期锁文档
lock_collection.create_index('expireAt', expireAfterSeconds=86400)

内容的提问来源于stack exchange，提问作者Tikrong