如何通过MongoDB聚合更新集合并将更新文档存入审计新集合？

你的核心需求是批量更新指定集合文档并同步写入审计集合，当前方案因聚合管道无法包含多个$merge，采用客户端遍历聚合结果的方式，但存在两个明显问题：

1. 聚合管道的$project仅保留description字段，导致后续forEach中无法获取ack和title，审计插入逻辑完全失效；
2. 客户端遍历所有聚合结果会产生不必要的网络和内存开销，尤其在数据量较大时性能低下。

以下是三种更优的替代方案，适配不同场景需求：

方案一：聚合+临时集合（高效批量处理） #

通过一次聚合完成文档处理并写入临时集合，再从临时集合批量同步到原集合和审计集合，仅需扫描一次源集合，效率最高。

public static void main(String[] args) {
    String uri = "mongodb://localhost:27017/?retryWrites=false";
    ServerApi serverApi = ServerApi.builder()
            .version(ServerApiVersion.V1)
            .build();

    MongoClientSettings settings = MongoClientSettings.builder()
            .applyConnectionString(new ConnectionString(uri))
            .serverApi(serverApi)
            .build();

    try (MongoClient client = MongoClients.create(settings)) {
        MongoDatabase database = client.getDatabase("myDB");
        MongoCollection<Document> collection = database.getCollection("myCollection1");
        MongoCollection<Document> auditCollection = database.getCollection("auditCollection");
        MongoCollection<Document> tempCollection = database.getCollection("temp_processed");

        // 1. 处理文档并写入临时集合（包含所有更新后字段）
        List<Bson> processPipeline = Arrays.asList(
                new Document("$match", new Document("type", "city")),
                new Document("$lookup", new Document()
                        .append("from", "country")
                        .append("localField", "cId")
                        .append("foreignField", "_id")
                        .append("as", "countryData")),
                new Document("$unwind", new Document()
                        .append("path", "$countryData")
                        .append("preserveNullAndEmptyArrays", true)),
                new Document("$set", new Document()
                        .append("data.desc", "$countryData.description")
                        .append("ack", false)),
                new Document("$out", "temp_processed")
        );
        collection.aggregate(processPipeline).allowDiskUse(true).toCollection();

        // 2. 将临时集合数据合并到原集合
        tempCollection.aggregate(Arrays.asList(
                new Document("$merge", new Document()
                        .append("into", "myCollection1")
                        .append("on", "_id")
                        .append("whenMatched", "merge")
                )
        )).toCollection();

        // 3. 提取审计字段写入审计集合
        tempCollection.aggregate(Arrays.asList(
                new Document("$project", new Document()
                        .append("_id", 1)
                        .append("title", 1)
                ),
                new Document("$merge", new Document()
                        .append("into", "auditCollection")
                        .append("on", "_id")
                        .append("whenMatched", "replace")
                        .append("whenNotMatched", "insert")
                )
        )).toCollection();

        // 4. 清理临时集合
        tempCollection.drop();

    } catch (MongoException e) {
        System.out.println(e.getMessage());
    }
}

适用场景：数据量较大，追求批量处理效率，不需要严格原子性。

方案二：事务（强原子性保障） #

如果你的MongoDB是副本集/分片集群（支持事务），可以在事务中完成"更新原文档+写入审计文档"的原子操作，确保两种操作要么同时成功，要么同时回滚。

public static void main(String[] args) {
    String uri = "mongodb://localhost:27017/?retryWrites=false";
    ServerApi serverApi = ServerApi.builder()
            .version(ServerApiVersion.V1)
            .build();

    MongoClientSettings settings = MongoClientSettings.builder()
            .applyConnectionString(new ConnectionString(uri))
            .serverApi(serverApi)
            .build();

    try (MongoClient client = MongoClients.create(settings)) {
        MongoDatabase database = client.getDatabase("myDB");
        MongoCollection<Document> collection = database.getCollection("myCollection1");
        MongoCollection<Document> auditCollection = database.getCollection("auditCollection");
        MongoCollection<Document> countryCollection = database.getCollection("country");

        Bson filter = new Document("type", "city");

        // 启动事务
        try (ClientSession session = client.startSession()) {
            session.startTransaction();

            try {
                // 遍历需要更新的文档
                for (Document doc : collection.find(session, filter)) {
                    // 获取关联的country数据
                    Document country = countryCollection.find(session, new Document("_id", doc.get("cId"))).first();

                    // 更新原文档
                    collection.updateOne(session,
                            new Document("_id", doc.get("_id")),
                            new Document("$set", new Document()
                                    .append("data.desc", country != null ? country.get("description") : null)
                                    .append("ack", false)));

                    // 写入审计文档
                    auditCollection.insertOne(session,
                            new Document("_id", doc.get("_id"))
                                    .append("title", doc.get("title")));
                }

                session.commitTransaction();
            } catch (Exception e) {
                session.abortTransaction();
                throw e;
            }
        }

    } catch (MongoException e) {
        System.out.println(e.getMessage());
    }
}

适用场景：数据量较小，需要严格保证更新与审计操作的原子性。

方案三：Change Streams（异步审计） #

通过监听原集合的变更事件，异步将更新后的文档写入审计集合，不阻塞主更新流程，适合高并发场景。

public static void main(String[] args) {
    String uri = "mongodb://localhost:27017/?retryWrites=false";
    ServerApi serverApi = ServerApi.builder()
            .version(ServerApiVersion.V1)
            .build();

    MongoClientSettings settings = MongoClientSettings.builder()
            .applyConnectionString(new ConnectionString(uri))
            .serverApi(serverApi)
            .build();

    try (MongoClient client = MongoClients.create(settings)) {
        MongoDatabase database = client.getDatabase("myDB");
        MongoCollection<Document> collection = database.getCollection("myCollection1");
        MongoCollection<Document> auditCollection = database.getCollection("auditCollection");

        // 监听原集合的更新操作
        ChangeStreamIterable<Document> changeStream = collection.watch(Arrays.asList(
                new Document("$match", new Document("operationType", "update"))
        ));

        // 处理变更事件，写入审计集合
        changeStream.forEach((Consumer<ChangeStreamDocument<Document>>) event -> {
            Document updatedDoc = event.getFullDocument();
            if (updatedDoc != null && updatedDoc.containsKey("ack")) {
                auditCollection.insertOne(new Document("_id", updatedDoc.get("_id"))
                        .append("title", updatedDoc.get("title")));
            }
        });

    } catch (MongoException e) {
        System.out.println(e.getMessage());
    }
}

适用场景：可以接受审计数据异步写入，主更新流程需要高性能，MongoDB为副本集/分片集群。

内容的提问来源于stack exchange，提问作者Kaveen M.