反应式Spring Boot中ReactiveMongoTemplate批量更新实现方法问询

需求可行性结论 #

该需求完全可以实现。Spring Data MongoDB的反应式模块已经原生封装了MongoDB的updateMany能力，同时也支持更灵活的批量多条件更新（即不同匹配条件对应不同更新逻辑的批量操作），都可以做到仅向MongoDB集群发送单次命令完成全部更新操作，避免多次IO开销。

具体实现方案 #

场景1：所有待更新条目使用相同的更新规则 #

该场景是MongoDB原生updateMany的典型适用场景，所有匹配查询条件的文档应用同一份更新规则，可直接使用ReactiveMongoTemplate的updateMulti方法实现：

// 示例：批量关闭所有创建超过30分钟且未支付的订单
// 构造查询条件
Query query = new Query(Criteria.where("status").is("UNPAID")
        .and("createTime").lt(LocalDateTime.now().minusMinutes(30)));
// 构造统一更新逻辑
Update update = Update.update("status", "CLOSED")
        .set("closeReason", "超时未支付自动关闭");
// 执行单次updateMany命令，返回更新条目数
reactiveMongoTemplate.updateMulti(query, update, Order.class)
        .subscribe(updateCount -> log.info("共关闭{}条超时订单", updateCount));

该方法底层直接调用MongoDB原生updateMany命令，全程仅发送1次数据库请求。

场景2：不同待更新条目使用不同的更新规则 #

如果批量更新的每个条目匹配条件、更新逻辑均不相同，可使用bulkWrite能力，将所有更新操作打包为单次请求发送到数据库：

// 示例：批量更新不同ID订单的状态和备注
List<Pair<Query, Update>> updateOperations = new ArrayList<>();
// orderUpdateList为业务层传入的待更新数据集合
for (OrderUpdateDTO updateDTO : orderUpdateList) {
    // 每个条目独立构造匹配条件
    Query query = new Query(Criteria.where("_id").is(updateDTO.getOrderId()));
    // 每个条目独立构造更新逻辑
    Update update = Update.update("status", updateDTO.getNewStatus())
            .set("remark", updateDTO.getRemark());
    updateOperations.add(Pair.of(query, update));
}
// 执行单次批量更新命令
// BulkMode.UNORDERED：不保证执行顺序，执行效率更高，单条失败不影响其他操作
// BulkMode.ORDERED：按顺序执行，某条操作失败则终止后续所有操作
reactiveMongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, Order.class)
        .updateMulti(updateOperations)
        .execute()
        .subscribe(result -> log.info("批量更新完成，匹配数：{}，更新数：{}",
                result.getMatchedCount(), result.getModifiedCount()));

该方法底层调用MongoDB原生bulkWrite命令，性能比循环调用单条更新高1~2个数量级。

注意事项 #

• 单次批量操作的总数据大小不能超过MongoDB 16MB的BSON文档限制，更新量极大时建议拆分多批执行
• 如果需要事务保证，需先开启MongoDB副本集/分片集群的事务支持，在对应方法上添加@Transactional注解即可
• 反应式编程中不要自行调用阻塞方法，所有结果处理通过subscribe或上游响应式操作符完成

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