mongodb使用docker搭建replicaSet集群与变更监听(最新推荐)

#在所有主机都创建用于存储mongodb数据的文件夹
mkdir -p ~/mongo-data/{data,key,backup}
#设置key文件，用于在集群机器间互相访问，各主机的key需要保持一致
cd ~/mongo-data
#在某一节点创建key
openssl rand -base64 123 > key/mongo-rs.key
sudo chown 999 key/mongo-rs.key
#不能是755, 权限太大不行. 
sudo chmod 600 key/mongo-rs.key
#将key复制到他节点
scp key/mongo-rs.key root@192.168.1.12:/root/mongo-data/key

var config={
     _id:"haiyangReplset",
     members:[
         {_id:0,host:"192.168.1.11:27017"},
         {_id:1,host:"192.168.1.12:27017"},
]};
rs.initiate(config)

#查看副本集状态
rs.status()
#查看副本集配置
rs.conf()
#添加节点
rs.add( { host: "ip:port"} )
#删除节点
rs.remove('ip:port')

package io.github.puhaiyang;

import com.google.common.collect.Lists;
import com.mongodb.client.*;
import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Scanner;
import java.util.concurrent.CompletableFuture;

/**
 * @author puhaiyang
 * @since 2023/3/30 20:01
 * MongodbWatchTestMain
 */
public class MongodbWatchTestMain {

    public static void main(String[] args) throws Exception {
        String uri = "mongodb://admin:123456@192.168.1.11:27017,192.168.1.12:27017/?replicaSet=haiyangReplset";
        MongoClient mongoClient = MongoClients.create(uri);
        MongoDatabase mongoDatabase = mongoClient.getDatabase("my-test-db");
        String myTestCollectionName = "myTestCollection";
        //获取出collection
        MongoCollection<Document> mongoCollection = initCollection(mongoDatabase, myTestCollectionName);
        //进行watch
        CompletableFuture.runAsync(() -> {
            while (true) {
                List<Bson> pipeline = Lists.newArrayList(
                        Aggregates.match(Filters.in("ns.coll", myTestCollectionName)),
                        Aggregates.match(Filters.in("operationType", Arrays.asList("insert", "update", "replace", "delete")))
                );
                ChangeStreamIterable<Document> changeStream = mongoDatabase.watch(pipeline)
                        .fullDocument(FullDocument.UPDATE_LOOKUP)
                        .fullDocumentBeforeChange(FullDocumentBeforeChange.WHEN_AVAILABLE);

                changeStream.forEach(event -> {
                    String collectionName = Objects.requireNonNull(event.getNamespace()).getCollectionName();
                    System.out.println("--------> event:" + event.toString());
                });
            }
        });

        //数据变更测试
        {
            Thread.sleep(3_000);
            InsertOneResult insertResult = mongoCollection.insertOne(new Document("test", "sample movie document"));
            System.out.println("Success! Inserted document id: " + insertResult.getInsertedId());
            UpdateResult updateResult = mongoCollection.updateOne(new Document("test", "sample movie document"), Updates.set("field2", "sample movie document update"));
            System.out.println("Updated " + updateResult.getModifiedCount() + " document.");
            DeleteResult deleteResult = mongoCollection.deleteOne(new Document("field2", "sample movie document update"));
            System.out.println("Deleted " + deleteResult.getDeletedCount() + " document.");
        }

        new Scanner(System.in).next();
    }

    private static MongoCollection<Document> initCollection(MongoDatabase mongoDatabase, String myTestCollectionName) {
        ArrayList<Document> existsCollections = mongoDatabase.listCollections().into(new ArrayList<>());
        Optional<Document> existsCollInfoOpl = existsCollections.stream().filter(doc -> StringUtils.equals(myTestCollectionName, doc.getString("name"))).findFirst();
        existsCollInfoOpl.ifPresent(collInfo -> {
            //确保开启了changeStreamPreAndPost
            Document changeStreamPreAndPostImagesEnable = collInfo.get("options", Document.class).get("changeStreamPreAndPostImages", Document.class);
            if (changeStreamPreAndPostImagesEnable != null && !changeStreamPreAndPostImagesEnable.getBoolean("enabled")) {
                Document mod = new Document();
                mod.put("collMod", myTestCollectionName);
                mod.put("changeStreamPreAndPostImages", new Document("enabled", true));
                mongoDatabase.runCommand(mod);
            }
        });
        if (!existsCollInfoOpl.isPresent()) {
            CreateCollectionOptions collectionOptions = new CreateCollectionOptions();
            //创建collection时开启ChangeStreamPreAndPostImages
            collectionOptions.changeStreamPreAndPostImagesOptions(new ChangeStreamPreAndPostImagesOptions(true));
            mongoDatabase.createCollection(myTestCollectionName, collectionOptions);
        }
        return mongoDatabase.getCollection(myTestCollectionName);
    }
}