服务器之家:专注于服务器技术及软件下载分享
分类导航

Mysql|Sql Server|Oracle|Redis|MongoDB|PostgreSQL|Sqlite|DB2|mariadb|Access|数据库技术|

服务器之家 - 数据库 - 数据库技术 - springboot整合mongodb changestream的示例代码

springboot整合mongodb changestream的示例代码

2023-05-07 12:12未知服务器之家 数据库技术

前言 changestream是monggodb的3.6版本之后出现的一种基于collection(数据库集合)的变更事件流,应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更 想必对mysql主从复制原理比较熟悉的同学应该知道,其根本就是从节

前言

changestream是monggodb的3.6版本之后出现的一种基于collection(数据库集合)的变更事件流,应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更

想必对mysql主从复制原理比较熟悉的同学应该知道,其根本就是从节点通过监听binlog日志,然后解析binlog日志数据达到数据同步的目的,于是,基于mysql主从复制原理,阿里开源了canal这样的数据同步中间件工具

Change Stream 介绍

Chang Stream(变更记录流) 是指collection(数据库集合)的变更事件流,应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更。

关于changestream做如下说明,提供参考

  • 在该特性出现之前,开发者可通过拉取 oplog达到同样的目的;
  • 但 oplog 的处理及解析相对复杂,而且存在被回滚的风险,如果使用不当的话还会带来性能问题;
  • Change Stream 可以与aggregate framework结合使用,对变更集进行进一步的过滤或转换;
  • 由于Change Stream 利用了存储在 oplog 中的信息,因此对于单进程部署的MongoDB无法支持Change Stream功能,其只能用于启用了副本集的独立集群或分片集群

changestream可用于监听的mongodb目标类型

  • 单个集合,除系统库(admin/local/config)之外的集合,3.6版本支持
  • 单个数据库,除系统库(admin/local/config)之外的数据库集合,4.0版本支持
  • 整个集群,整个集群内除去系统库( (admin/local/config)之外的集合 ,4.0版本支持

一个Change Stream Event的基本结构如下所示:

{
   _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>
   }
}

关于上面的数据结构,做简单的解释说明,

  • _id,变更事件的Token对象
  • operationType,变更类型(见下面介绍)
  • fullDocument,文档内容
  • ns,监听的目标
  • ns.db,变更的数据库
  • ns.coll,变更的集合
  • documentKey,变更文档的键值,含_id字段
  • updateDescription,变更描述
  • updateDescription.updatedFields,变更中更新字段
  • updateDescription.removedFields,变更中删除字段
  • clusterTime,对应oplog的时间戳
  • txnNumber,事务编号,仅在多文档事务中出现,4.0版本支持
  • lsid,事务关联的会话编号,仅在多文档事务中出现,4.0版本支持

Change Steram支持的变更类型,对于上面的operationType 这个参数,主要包括有以下几个:

  • insert,插入文档
  • delete,删除文档
  • replace,替换文档,当执行replace操作指定upsert时,可能是insert事件
  • update,更新文档,当执行update操作指定upsert时,可能是insert事件
  • invalidate,失效事件,比如执行了collection.drop或collection.rename

以上的几种类型,可以简单理解为,监听的mongo用户操作的事件类型,比如新增数据,删除数据,修改数据等

以上为changestream的必备理论知识,想要深入学习的话无比要了解,下面通过实操来展示下changestream的使用

环境准备

mongdb复制集群,本例的复制集群对应的mongodb版本为 4.0.X

springboot整合mongodb changestream的示例代码

登录primary节点,创建一个数据库

springboot整合mongodb changestream的示例代码

友情提醒:数据库需要提前创建

1、启动两个Mongo shell,一个操作数据库,一个watch

在其中一个窗口执行如下命令,开启监听

cursor = db.comment.watch()

springboot整合mongodb changestream的示例代码

2、在另一个窗口下,给上面的articledb插入一条数据

springboot整合mongodb changestream的示例代码

数据写入成功后,在第一个窗口下,执行下面的命令:

cursor.next()

springboot整合mongodb changestream的示例代码

说明已经成功监听到新增的数据,修改、删除事件可以做类似的操作即可

以上先通过shell窗口展示了一下changestream的使用效果,接下来,将通过程序演示下如何在客户端集成并使用changestream

Java客户端操作changestream

1、引入maven依赖

<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>

2、测试类核心代码

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);
        }
}

这段程序主要分为几个核心部分,做如下解释说明,

  • 连接mogodb服务端及相关配置
  • 通过pipline开启watch监听
  • 监听到特定数据库下集合的数据变化,然后打印出变化的数据

启动这段程序,观察控制台日志数据

springboot整合mongodb changestream的示例代码

在未对articledb数据库下的comment集合做任何操作之前,由于watch为检测到任何数据变化,所以无法进入到while循环中,接下来,从shell端给comment集合新增一条数据,然后再次观察控制台数据变化

springboot整合mongodb changestream的示例代码

可以看到,控制台很快就检测到变化的数据

springboot整合mongodb changestream的示例代码

以下为完整的日志数据

{ operationType=OperationType{value='insert'}, resumeToken={"_data": "8262138891000000022B022C0100296E5A1004B9065629412942F8852D592B9FD441B946645F696400646213889158B116A29C3FD1140004"}, namespace=articledb.comment, destinationNamespace=null, fullDocument=Document{{_id=6213889158b116a29c3fd114, articleid=100010, content=hello kafka, userid=1010, nickname=marry}}, documentKey={"_id": {"$oid": "6213889158b116a29c3fd114"}}, clusterTime=Timestamp{value=7067142396626075650, seconds=1645447313, inc=2}, updateDescription=null, txnNumber=null, lsid=null}

至于在业务中的具体使用,可以结合自身的情况,举例来说,应用程序只想监听修改数据的事件,那么就可以在修改数据事件的监听逻辑中,解析变化后的数据做后续的操作

springboot整合changestream

在实际开发中,更通用的场景是整合到springboot工程中使用,有过一定的开发经验的同学应该很容易想到核心的逻辑长什么样了,和canal的客户端操作类似,需要在一个配置类去监听即可

下面来看看具体的整合步骤

1、引入核心依赖

<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>

2、核心配置文件

本例演示的是基于上文搭建的mongodb复制集群

server.port=8081

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

3、编写实体类,映射comment集合中的字段

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;
}

4、编写一个服务类

简单的添加2个用接口测试的方法

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;
    /**
     * 查询所有
     * @return
     */
    public List<Comment> findAll() {
        return mongoTemplate.findAll(Comment.class);
    }
    /***
     * 根据id查询
     * @param id
    public Comment getBookById(String id) {
        Query query = new Query(Criteria.where("articleid").is(id));
        return mongoTemplate.findOne(query, Comment.class);
}

5、编写一个接口

@RestController
public class CommentController {

    @Autowired
    private MongoDbService mongoDbService;
    @GetMapping("/listAll")
    public Object listAll(){
        return mongoDbService.findAll();
    }
    @GetMapping("/findById")
    public Object findById(String id){
        return mongoDbService.getBookById(id);
}

启动本工程,然后浏览器调用下查询所有数据的接口,数据能正常返回,说明工程的基础结构就完成了

springboot整合mongodb changestream的示例代码

6、接下来,只需要依次添加下面3个配置类即可

MongoMessageListener 类 ,顾名思义,该类用于监听特定数据库下的集合数据变化使用的,在实际开发中,该类的作用也是非常重要的,类似于许多中间件的客户端监听程序,当监听到数据变化后,做出后续的业务响应,比如,数据入库、推送消息到kafka、发送相关的事件等等

import com.congge.entity.Comment;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.OperationType;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListener;
import org.springframework.stereotype.Component;

@Component
public class MongoMessageListener implements MessageListener<ChangeStreamDocument<Document>,Comment> {
    private static Logger logger = LoggerFactory.getLogger(MongoMessageListener.class);
    @Override
    public void onMessage(Message<ChangeStreamDocument<Document>, Comment> message) {
        OperationType operationType = message.getRaw().getOperationType();
        System.out.println("操作类型为 :" + operationType);
        System.out.println("变更数据主体 :" + message.getBody().getArticleid());
        System.out.println("变更数据主体 :" + message.getBody().getContent());
        System.out.println("变更数据主体 :" + message.getBody().getNickname());
        System.out.println("变更数据主体 :" + message.getBody().getUserid());
        System.out.println();
        /*logger.info("Received Message in collection: {},message raw: {}, message body:{}",
                message.getProperties().getCollectionName(), message.getRaw(), message.getBody());*/
    }
}

ChangeStream 类 ,事件注册类,即开篇中提到的那几种事件类型的操作等

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);
    }
}

MongoConfig 配置MessageListenerContainer 容器的相关参数

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;
            }
        };
    }
}

3个类添加完成后,再次启动程序,并观察控制台数据日志

测试1:通过shell窗口登录primary节点,并给comment集合添加一条数据

springboot整合mongodb changestream的示例代码

几乎是实时的监听到事件操作的数据变化,下面是完整的输出日志

测试2:通过shell窗口删除上面新增的这条数据

springboot整合mongodb changestream的示例代码

典型应用场景

数据迁移

如果一个系统的数据需要迁移到另一个系统,可以考虑使用mongodb changestream这种方式,试想,如果老系统数据非常杂乱,并且文档中存在一些脏数据时,为了确保迁移后的数据能较快的投产,通过应用程序的方式,能够原始的数据做类似ETL的处理,这样更加方便

应用监控

如果您的系统对数据监管较为严格,可以考虑使用changestream这种方式,订阅特定事件的数据操作,比如修改和删除数据的事件,然后及时的发送告警通知

对接大数据应用

我们知道,mongodb作为一款性能优秀的分布式文档型数据库,其实是可以存储海量数据的,在一些大数据场景下,比如下游其他的应用采用大数据技术,需要对mongo中的数据做轨迹行为分析,changestream就是一种不错的选择,当监听到特定事件的数据变化时,向消息队列,比如kafka推送相应的消息,下游相关的大数据应用就可以做后续的业务处理了

延伸 · 阅读

精彩推荐