基于Reactive Mongo实现嵌套对象列表的Upsert操作

先纠正实体模型的误区 #

你的OrderItem和Supplier类上标注了@Document，这会让它们被当作独立的Mongo集合实体，但当前你是把它们作为Order的嵌套子文档使用，这会导致序列化/反序列化异常，也不符合嵌套文档的设计。需要移除这两个类上的@Document注解：

修改后的OrderItem：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;

import java.math.BigDecimal;
import java.util.List;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderItem {
   @Id
   String orderItemId;
   BigDecimal price;
   String partNumber;
   String description;
   int quantityInStock;
   List<Supplier> suppliers;
}

修改后的Supplier：

import lombok.Data;
import org.springframework.data.annotation.Id;

@Data
public class Supplier {
   @Id
   String supplierId;
   String supplierName;
   String supplierPartNumber;
}

嵌套集合的Upsert实现 #

根据你的需求，分两种常见场景实现：

场景1：直接替换整个OrderItem列表 #

如果每次upsert时需要完全覆盖订单下的所有OrderItem，直接用update.set()设置整个orderItemList即可，写法最简单：

protected Update createOrderUpdate(Order order) {
    Update update = new Update();
    // @Id对应Mongo的_id字段，orderId与_id自动映射，无需单独设置
    update.set("orderItemList", order.getOrderItemList());
    return update;
}

适配Reactive Mongo的批量upsert逻辑（替换原同步MongoOperations为Reactive版本）：

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
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 org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderService {

    private final ReactiveMongoOperations reactiveMongoOperations;
    private final static String ORDER_COLLECTION = "orders";

    public Mono<Void> upsertOrders(List<Order> orders) {
        if (orders.isEmpty()) {
            return Mono.empty();
        }

        return Flux.fromIterable(orders)
                .flatMap(order -> {
                    Query query = Query.query(Criteria.where("_id").is(order.getOrderId()));
                    Update update = createOrderUpdate(order);
                    return reactiveMongoOperations.upsert(query, update, ORDER_COLLECTION);
                })
                .then()
                .onErrorMap(e -> {
                    log.error("Failed to upsert orders: {}", e.getMessage());
                    return e;
                });
    }

    protected Update createOrderUpdate(Order order) {
        Update update = new Update();
        update.set("orderItemList", order.getOrderItemList());
        return update;
    }
}

场景2：根据OrderItemId匹配更新/新增单个元素 #

如果需要对OrderItem列表做精细化更新（存在则更新，不存在则追加），可以使用Mongo的数组过滤更新，结合Spring Data Mongo的Update.filterArray()方法实现：

protected Update createOrderUpdate(Order order) {
    Update update = new Update();

    // 遍历每个OrderItem，分别处理匹配更新与追加逻辑
    for (OrderItem item : order.getOrderItemList()) {
        // 更新已存在的OrderItem（匹配orderItemId）
        update.filterArray(Criteria.where("orderItemId").is(item.getOrderItemId()))
                .set("orderItemList.$", item);

        // 追加不存在的OrderItem（addToSet自动去重，若需强制重复追加改用push）
        update.addToSet("orderItemList", item);
    }

    return update;
}

批量处理的Reactive逻辑与场景1一致，仅需替换createOrderUpdate方法即可。

关键说明 #

1. Reactive适配：Reactive环境下必须使用ReactiveMongoOperations，所有操作返回Mono/Flux，避免阻塞。
2. @Id映射：Order类的orderId标注@Id后，会自动映射到Mongo文档的_id字段，upsert查询条件直接用_id匹配即可。
3. 嵌套文档序列化：移除@Document后的OrderItem和Supplier会被当作普通POJO，自动序列化为Order文档的嵌套子文档，存储在orderItemList数组中。

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