如何使用Spring/Java实现MongoDB原子性读取与更新操作

MongoDB单文档级别的读写操作天生具备原子性，不需要额外引入分布式锁组件，直接使用findAndModify命令即可实现「查询待处理任务+标记为运行中」的全流程原子性，完全避免多实例争抢同一个任务的问题。

前置约定 #

首先给任务集合定义统一的状态字段规则：

• 任务基础字段包含：主键_id、任务状态status、计划执行时间executeTime、拾取实例标识instanceId、状态更新时间updateTime、优先级priority、创建时间createTime
• 状态枚举值：pending（待处理）、running（处理中）、success（执行成功）、failed（执行失败）

核心实现（Spring Data MongoDB版） #

直接使用MongoTemplate提供的findAndModify方法即可，注意不要把查询和更新拆成两个独立操作，否则会出现并发窗口。

import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import java.time.LocalDateTime;

// 直接注入即可使用
private final MongoTemplate mongoTemplate;

/**
 * 原子拾取待处理任务
 * @param currentInstanceId 当前运行实例的唯一标识，可使用IP+端口/UUID生成
 * @return 拾取到的任务，无待处理任务时返回null
 */
public Task pickPendingTask(String currentInstanceId) {
    // 构造查询条件：筛选待处理、已到执行时间的任务
    Query query = Query.query(
            Criteria.where("status").is("pending")
                    .and("executeTime").lte(LocalDateTime.now())
        )
        // 按优先级、创建时间升序，优先处理高优先级、早创建的任务
        .with(Sort.by(Sort.Direction.ASC, "priority", "createTime"))
        // 每次只拾取1条任务
        .limit(1);

    // 构造原子更新逻辑：将匹配到的任务标记为运行中，记录拾取实例和时间
    Update update = new Update()
            .set("status", "running")
            .set("instanceId", currentInstanceId)
            .set("updateTime", LocalDateTime.now());

    // 配置操作选项：返回修改前的文档，不做upsert
    FindAndModifyOptions options = FindAndModifyOptions.options()
            .returnNew(false)
            .upsert(false);

    // 操作在MongoDB服务端原子执行：匹配文档→修改文档→返回文档全流程不会被其他实例的请求打断
    return mongoTemplate.findAndModify(query, update, options, Task.class);
}

生产环境注意事项 #

• 禁止拆分查询和更新步骤：不要先查询status为pending的任务列表，再单独调用update修改状态，两个操作之间存在并发间隙，会导致多个实例拿到同一个任务。
• 索引优化：给status+executeTime字段创建联合索引，大幅提升任务拾取的查询效率，避免大集合下的全表扫描。
• 超时兜底逻辑：增加定时巡检任务，把处于running状态超过阈值（比如30分钟，根据业务实际执行时长配置）的任务重置为pending状态，避免实例宕机导致任务永远卡在运行中无法被执行。
• 批量拾取场景：如果需要单次拾取多条任务，不要直接用updateMulti，可以循环调用上述单条拾取方法，或者使用MongoDB 4.2+版本支持的聚合管道更新，确保每条任务只会被一个实例匹配修改成功。

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