springboot整合mongodb changestream代码

package com.congge.test; import com.mongodb.*; import com.mongodb.client.MongoDatabase; import org.bson.conversions.Bson; import java.util.List; import static java.util.Collections.singletonList; import com.alibaba.fastjson.JSONObject; import com.mongodb.MongoClient; import com.mongodb.MongoClientURI; import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoCursor; import com.mongodb.client.MongoDatabase; import com.mongodb.client.model.Aggregates; import com.mongodb.client.model.Filters; import com.mongodb.client.model.changestream.ChangeStreamDocument; import org.bson.Document; import org.bson.conversions.Bson; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import static java.util.Arrays.asList; public class MongoTest { private static Logger logger = LoggerFactory.getLogger(MongoTest.class); public static void main(String[] args) { /*MongoClient mongoClient = new MongoClient("IP", 27017); DB db = mongoClient.getDB("articledb"); DBCollection collection = db.getCollection("comment"); BasicDBObject query = new BasicDBObject("articleid", "100007"); DBCursor cursor = collection.find(query); try { while(cursor.hasNext()) { System.out.println(cursor.next()); } } finally { cursor.close(); }*/ mogodbdata(); } private static void mogodbdata() { String sURI = "mongodb://IP:27017"; MongoClient mongoClient = new MongoClient(new MongoClientURI(sURI)); MongoDatabase database = mongoClient.getDatabase("articledb"); MongoCollection<Document> collec = database.getCollection("comment"); List<Bson> pipeline = singletonList(Aggregates.match(Filters.or( Document.parse("{'fullDocument.articleid': '100007'}"), Filters.in("operationType", asList("insert", "update", "delete"))))); MongoCursor<ChangeStreamDocument<Document>> cursor = collec.watch(pipeline).iterator(); while (cursor.hasNext()) { ChangeStreamDocument<Document> next = cursor.next(); logger.info("输出mogodb的next的对应的值" + next.toString()); String Operation = next.getOperationType().getValue(); String tableNames = next.getNamespace().getCollectionName(); System.out.println(tableNames); //获取主键id的值 String pk_id = next.getDocumentKey().toString(); //同步修改数据的操作 if (next.getUpdateDescription() != null) { JSONObject jsonObject = JSONObject.parseObject(next.getUpdateDescription().getUpdatedFields().toJson()); System.out.println(jsonObject); } //同步插入数据的操作 if (next.getFullDocument() != null) { JSONObject jsonObject = JSONObject.parseObject(next.getFullDocument().toJson()); System.out.println(jsonObject); } //同步删除数据的操作 if (next.getUpdateDescription() == null && Operation.matches("delete")) { JSONObject jsonObject = JSONObject.parseObject(pk_id); System.out.println(jsonObject); } } } }