MongoDB变更流聚合投影嵌套变量：Java语法适配求助

以下是将你在Compass中验证通过的聚合管道转换为Java Change Streams语法的代码，完全满足你的需求：筛选特定_id的replace类型变更事件，并仅保留Members数组中State为true的元素。

完整代码示例 #

import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Projections;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import org.bson.Document;
import java.util.Arrays;

import static com.mongodb.client.model.Filters.eq;
import static com.mongodb.client.model.changestream.FullDocument.UPDATE_LOOKUP;

public class MongoChangeStreamDemo {
    public static void main(String[] args) {
        // 连接MongoDB
        try (var mongoClient = MongoClients.create("mongodb://localhost:27017")) {
            MongoDatabase db = mongoClient.getDatabase("你的数据库名");
            MongoCollection<Document> collection = db.getCollection("你的集合名");

            // 构建聚合管道，对应Compass中的逻辑
            var pipeline = Arrays.asList(
                // 匹配条件：replace事件 + 目标_id的文档
                Aggregates.match(Filters.and(
                    eq("operationType", "replace"),
                    eq("fullDocument._id", "Omega")
                )),
                // 投影并过滤Members数组
                Aggregates.project(Projections.fields(
                    Projections.computed("Members",
                        Document.parse("""
                            {
                                $filter: {
                                    input: "$fullDocument.Members",
                                    as: "item",
                                    cond: { $eq: ["$$item.State", true] }
                                }
                            }
                        """)
                    ),
                    // 可选保留变更流元数据
                    Projections.include("operationType", "clusterTime")
                ))
            );

            // 启动变更流监听
            collection.watch(pipeline)
                     .fullDocument(UPDATE_LOOKUP) // 获取更新后的完整文档
                     .forEach(changeStreamDoc -> {
                         System.out.println("捕获到符合条件的变更：");
                         System.out.println(changeStreamDoc.toJson());
                     });
        }
    }
}

核心要点说明 #

1. 变更事件类型过滤：新增operationType: "replace"的匹配，确保只监听你需要的replace类型变更。
2. fullDocument字段：变更流中，更新后的完整文档存储在fullDocument字段下，所以管道里的字段路径要从$Members改为$fullDocument.Members，这是和直接集合聚合的关键差异。
3. 复用Filter逻辑：直接用Document.parse()导入你在Compass中验证过的$filter表达式，避免手动拼接出错。
4. 依赖要求：确保项目使用MongoDB Java Driver 4.x及以上版本，Maven依赖示例：

<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongodb-driver-sync</artifactId>
    <version>4.11.1</version>
</dependency>

内容的提问来源于stack exchange，提问作者K-ous