springboot整合mongodb？changestream的示例代码（springboot整合mongodb查询嵌套数据）难以置信

目录前言Change Stream 介绍环境准备Java客户端操作changestream1、引入maven依赖2、测试类核心代码下面来看看具体的整合步骤1、引入核心依赖2、核心配置文件3、编写实体类，映射comment集合中的字段4、编写一个服务类5、编写一个接口6、接下来，只需要依次添加下面3个配置类即可典型应用场景数据迁移应用监控对接大数据应用

{
_id : { <BSON Object> },
“operationType” : “<operation>”,
“fullDocument” : { <document> },
“ns” : {
“db” : “<database>”,
“coll” : “<collection”
},
“documentKey” : { “_id” : <ObjectId> },
“updateDescription” : {
“updatedFields” : { <document> },
“removedFields” : [ “<field>”, … ]
}
“clusterTime” : <Timestamp>,
“txnNumber” : <NumberLong>,
“lsid” : {
“id” : <UUID>,
“uid” : <BinData>
}
}

cursor=db.comment.watch()

cursor.next()

<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongo-java-driver</artifactId>
<version>3.12.2</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.75</version>
</dependency>

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.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.util.ArrayList;
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) {
showmogodbdata();
}
private static void showmogodbdata() {

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());
//同步删除数据的操作
if (next.getUpdateDescription()==null && Operation.matches(“delete”)) {
JSONObject jsonObject=JSONObject.parseObject(pk_id);
}
}

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<artifactId>spring-boot-starter</artifactId>

server.port=8081

#mongodb配置
spring.data.mongodb.uri=mongodb://IP:27017,IP:27018,IP:27019/articledb?maxPoolSize=512

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;

@Document(collection=”comment”)
public class Comment {
@Id
private String articleid;
private String content;
private String userid;
private String nickname;
private Date createdatetime;
public String getArticleid() {
return articleid;
}
public void setArticleid(String articleid) {
this.articleid=articleid;
public String getContent() {
return content;
public void setContent(String content) {
this.content=content;
public String getUserid() {
return userid;
public void setUserid(String userid) {
this.userid=userid;
public String getNickname() {
return nickname;
public void setNickname(String nickname) {
this.nickname=nickname;
public Date getCreatedatetime() {
return createdatetime;
public void setCreatedatetime(Date createdatetime) {
this.createdatetime=createdatetime;
}

import com.congge.entity.Comment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import java.util.List;
@Service
public class MongoDbService {
private static final Logger logger=LoggerFactory.getLogger(MongoDbService.class);
@Autowired
private MongoTemplate mongoTemplate;

public List<Comment> findAll() {
return mongoTemplate.findAll(Comment.class);
}

}
}

import com.congge.entity.Comment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.query.Criteria;

@Configuration
public class ChangeStream implements CommandLineRunner {
@Autowired
private MongoMessageListener mongoMessageListener;
private MessageListenerContainer messageListenerContainer;
@Override
public void run(String… args) throws Exception{
ChangeStreamRequest<Comment> request=ChangeStreamRequest.builder(mongoMessageListener)
.collection(“comment”)
.filter(Aggregation.newAggregation(Aggregation.match(Criteria.where(“operationType”).in(“insert”,”update”,”replace”))))
.build();
messageListenerContainer.register(request,Comment.class);
}
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
@Configuration
public class MongoConfig {
@Bean
MessageListenerContainer messageListenerContainer(MongoTemplate mongoTemplate){
Executor executor=Executors.newFixedThreadPool(5);
return new DefaultMessageListenerContainer(mongoTemplate,executor){
@Override
public boolean isAutoStartup(){
return true;
}
};
}
}

您可能感兴趣的文章:SpringBoot整合redis及mongodb的详细过程SpringBoot整合Mongodb实现增删查改的方法SpringBoot?整合mongoDB并自定义连接池的示例代码SpringBoot整合MongoDB完整实例代码SpringBoot整合之SpringBoot整合MongoDB的详细步骤SpringBoot MongoDB详细使用教程