使用ReactiveMongo按文档内条件更新并返回更新后文档的问题

你不需要拆分操作，ReactiveMongo（1.1.x及以上版本）支持通过聚合管道执行findOneAndUpdate，同时返回更新后的文档。以下是可行的实现方案：

核心实现代码 #

首先确保你的项目依赖了ReactiveMongo 1.1.x或更高版本，然后编写如下代码：

import reactivemongo.api.bson._
import reactivemongo.api.bson.collection.BSONCollection
import scala.concurrent.{ExecutionContext, Future}
import java.time.Instant

// 锁定时间配置类，根据实际需求调整字段
case class TaskLockConfig(
  retry1LockUntil: BSONDateTime,
  retry2LockUntil: BSONDateTime,
  retry3LockUntil: BSONDateTime,
  // ... 补充retry4到retry10的配置项
  defaultLockUntil: BSONDateTime
)

def acquireNextReadyTask(
  collection: BSONCollection,
  lockConfig: TaskLockConfig
)(implicit ec: ExecutionContext): Future[Option[BSONDocument]] = {
  val currentTime = BSONDateTime(Instant.now().toEpochMilli)

  // 1. 构建查询条件：匹配READY状态且未锁定的任务
  val query = BSONDocument(
    "status" -> "READY",
    "lockedUntil" -> BSONDocument("$lt" -> currentTime)
  )

  // 2. 构建聚合管道更新逻辑：根据retries值设置lockedUntil
  val updatePipeline = List(
    BSONDocument(
      "$set" -> BSONDocument(
        "lockedUntil" -> BSONDocument(
          "$switch" -> BSONDocument(
            "branches" -> List(
              BSONDocument(
                "case" -> BSONDocument("$eq" -> List("$retries", 1)),
                "then" -> lockConfig.retry1LockUntil
              ),
              BSONDocument(
                "case" -> BSONDocument("$eq" -> List("$retries", 2)),
                "then" -> lockConfig.retry2LockUntil
              ),
              BSONDocument(
                "case" -> BSONDocument("$eq" -> List("$retries", 3)),
                "then" -> lockConfig.retry3LockUntil
              ),
              // ... 依次添加retries 4到10的分支
            ),
            "default" -> lockConfig.defaultLockUntil
          )
        )
      )
    )
  )

  // 3. 执行findAndUpdate，返回更新后的文档
  collection.findAndUpdate(
    selector = query,
    update = updatePipeline, // 传入聚合管道列表
    fetchNewObject = true,   // 关键：返回更新后的文档
    upsert = false,
    sort = Some(BSONDocument("retries" -> 1)), // 可选：优先处理重试次数少的任务
    bypassDocumentValidation = false
  ).map(_.result)
}

关键说明 #

1. 版本要求：ReactiveMongo 1.1.x及以上才支持将聚合管道作为findAndUpdate的更新参数，旧版本会报错，务必升级依赖。
2. 原子性保障：findAndUpdate是原子操作，避免了"查询-更新"拆分带来的并发冲突问题。
3. 返回更新后文档：通过fetchNewObject = true配置，直接获取更新后的完整文档，无需额外查询。
4. BSON类型匹配：确保lockedUntil的配置值是BSONDateTime类型，与数据库中的ISO时间字段类型一致，避免类型错误。

解决你遇到的问题 #

• 之前findAndUpdate不支持聚合管道：大概率是版本过低，升级到1.1.x+即可解决。
• update.one无法返回文档：确实如此，update系列方法只返回更新结果统计，不返回文档，必须用findAndUpdate并开启fetchNewObject。

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