如何在Spring Data ChangeStreamOptions.filter中传入MongoDB聚合JSON字符串

我明白你想把ChangeStreamOptions的filter从基于Criteria的Aggregation改成直接使用MongoDB聚合JSON字符串的需求，之前的写法出错主要是因为对Document解析和ChangeStream聚合结构的处理不对，下面给你两种可行的解决方案：

方案一：直接解析JSON聚合数组构建Aggregation #

ChangeStream的filter接受的Aggregation本质是由多个管道步骤组成的列表，你不能直接用Document.parse()解析数组字符串（因为Document只能表示单个BSON对象），需要先把数组包裹在一个临时对象里再提取列表：

import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.ChangeStreamOptions;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import java.util.List;

@Service
public class Watch {

    private static final Logger log = LoggerFactory.getLogger(Watch.class);
    @Autowired
    private ReactiveMongoTemplate reactiveMongoTemplate;

    public Flux<Person> watchAgeGreaterThan(Integer ageParam) {
        // 构建符合ChangeStream结构的聚合JSON，注意要匹配operationType和fullDocument下的age字段
        String aggregationJson = String.format(
                "[{\"$match\": {\"operationType\": \"insert\", \"fullDocument.age\": {\"$gt\": %d}}}",
                ageParam
        );

        // 解析JSON数组：先把数组包裹在临时对象中，再提取管道步骤列表
        List<Document> pipelineStages = Document.parse("{\"stages\": " + aggregationJson + "}")
                .getList("stages", Document.class);

        // 用管道列表构建Aggregation
        Aggregation aggregation = Aggregation.newAggregation(pipelineStages);

        ChangeStreamOptions options = ChangeStreamOptions.builder()
                .filter(aggregation)
                .returnFullDocumentOnUpdate()
                .build();

        return reactiveMongoTemplate.changeStream("person_collection", options, Person.class)
                .map(ChangeStreamEvent::getBody)
                .doOnError(throwable -> log.error("Error on 'person' change stream event :: " + throwable.getMessage(), throwable));
    }
}

方案二：自定义AggregationOperation处理单条JSON管道步骤 #

如果你想拆分多个管道步骤为单独的JSON字符串（比如把operationType匹配和age匹配分开），可以修正你之前的自定义AggregationOperation实现，确保每个JSON对应一个管道步骤：

第一步：实现通用的JSON聚合操作类 #

import org.bson.Document;
import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;

public class JsonPipelineOperation implements AggregationOperation {
    private final Document pipelineDocument;

    public JsonPipelineOperation(String pipelineJson) {
        this.pipelineDocument = Document.parse(pipelineJson);
    }

    @Override
    public Document toDocument(AggregationOperationContext context) {
        // 利用上下文自动映射实体类字段（比如Person类用@Field注解的字段会被正确转换）
        return context.getMappedObject(pipelineDocument);
    }
}

第二步：在Watch服务中使用自定义操作 #

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.ChangeStreamOptions;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

@Service
public class Watch {

    private static final Logger log = LoggerFactory.getLogger(Watch.class);
    @Autowired
    private ReactiveMongoTemplate reactiveMongoTemplate;

    public Flux<Person> watchAgeGreaterThan(Integer ageParam) {
        // 定义两个独立的管道步骤JSON
        String matchOperationType = "{\"$match\": {\"operationType\": \"insert\"}}";
        String matchAge = String.format("{\"$match\": {\"fullDocument.age\": {\"$gt\": %d}}}", ageParam);

        // 用自定义操作构建Aggregation
        Aggregation aggregation = Aggregation.newAggregation(
                new JsonPipelineOperation(matchOperationType),
                new JsonPipelineOperation(matchAge)
        );

        ChangeStreamOptions options = ChangeStreamOptions.builder()
                .filter(aggregation)
                .returnFullDocumentOnUpdate()
                .build();

        return reactiveMongoTemplate.changeStream("person_collection", options, Person.class)
                .map(ChangeStreamEvent::getBody)
                .doOnError(throwable -> log.error("Error on 'person' change stream event :: " + throwable.getMessage(), throwable));
    }
}

关键注意点 #

• ChangeStream事件的结构中，插入的文档数据在fullDocument字段下，所以你的聚合JSON中必须用fullDocument.age来匹配字段，而不能直接写age。
• 如果需要动态参数（比如ageParam），尽量用字符串格式化或者参数绑定的方式，避免手动拼接JSON时出现语法错误或安全问题。
• 自定义AggregationOperation中的context.getMappedObject()方法很重要，它会帮你处理实体类和MongoDB集合字段的映射关系（比如使用@Field注解的字段）。

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