我如何在 Spring Boot 中迭代 MongoDB 更改流？答案

【问题标题】：How do I iterate over a MongoDB Change Stream in Spring Boot?我如何在 Spring Boot 中迭代 MongoDB 更改流？
【发布时间】：2022-12-07 17:23:04
【问题描述】：

我已经阅读了无数关于 MongoDB Change Streams 的文章和代码示例，但我仍然无法正确设置它。我正在尝试收听我的 MongoDB 中的特定集合，每当插入、更新或删除文档时，我都想做点什么。

这是我试过的：

@Data
@Document(collection = "teams")
public class Teams{
    private @MongoId(FieldType.OBJECT_ID)
    ObjectId id;
    private Integer teamId;
    private String name;
    private String description;
}

import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoClient;
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.changestream.FullDocument;
import com.mongodb.client.ChangeStreamIterable;

import org.bson.Document;
import org.bson.conversions.Bson;

import java.util.Arrays;
import java.util.List;

public class MongoDBChangeStream {

    // connect to the local database server
    MongoClient mongoClient = MongoClients.create("db uri goes here");

    // Select the MongoDB database
    MongoDatabase database = mongoClient.getDatabase("MyDatabase");

    // Select the collection to query
    MongoCollection<Document> collection = database.getCollection("teams");

    // Create pipeline for operationType filter
    List<Bson> pipeline = Arrays.asList(
            Aggregates.match(
                    Filters.in("operationType",
                            Arrays.asList("insert", "update", "delete"))));

    // Create the Change Stream
    ChangeStreamIterable<Document> changeStream = collection.watch(pipeline)
            .fullDocument(FullDocument.UPDATE_LOOKUP);

    // Iterate over the Change Stream
    for (Document changeEvent : changeStream) {
        // Process the change event here
    }
}

所以这就是我到目前为止所拥有的，一切都很好，直到出现三个错误的 for 循环：

下面有一条红线'为了（'，上面写着unexpected token。
下面有一条红线'：'，上面写着';' expected。
下面有一条红线'改变流）'，上面写着unknown class: 'changeStream'。

【问题讨论】：

标签： java mongodb spring-boot watch changestream

【解决方案1】：

首先，您应该将代码放在类方法中，而不是类主体中。第二 - ChangeStreamIterable<Document> 迭代器元素是 ChangeStreamDocument<Document> 而不是 Document。

总结一下：

public class MongoDBChangeStream {

    public void someMethod() {

        // connect to the local database server
        MongoClient mongoClient = MongoClients.create("db uri goes here");

        // Select the MongoDB database
        MongoDatabase database = mongoClient.getDatabase("MyDatabase");

        // Select the collection to query
        MongoCollection<Document> collection = database.getCollection("teams");

        // Create pipeline for operationType filter
        List<Bson> pipeline = Arrays.asList(
                Aggregates.match(
                        Filters.in(
                                "operationType",
                                Arrays.asList("insert", "update", "delete")
                        )));

        // Create the Change Stream
        ChangeStreamIterable<Document> changeStream = collection.watch(pipeline)
                .fullDocument(FullDocument.UPDATE_LOOKUP);

        // Iterate over the Change Stream
        for (ChangeStreamDocument<Document> changeEvent : changeStream) {
            // Process the change event here
        }
    }
}

【讨论】：

【解决方案2】：

迭代 changeStream 的 for 循环不在方法内。这会导致代码执行时出错，因为for循环除非在方法内部，否则无法执行。

【讨论】：