产品中心
changestream是monggodb的3.6版本之后出现的一种基于collection(数据库集合)的变更事件流，应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更
想必对mysql主从复制原理比较熟悉的同学应该知道，其根本就是从节点通过监听binlog日志，然后解析binlog日志数据达到数据同步的目的，于是，基于mysql主从复制原理，阿里开源了canal这样的数据同步中间件工具
Chang Stream(变更记录流) 是指collection(数据库集合)的变更事件流，应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更。
关于changestream做如下说明，提供参考
changestream可用于监听的mongodb目标类型
一个Change Stream Event的基本结构如下所示：
登录primary节点，创建一个数据库
友情提醒：数据库需要提前创建
1、启动两个Mongo shell，一个操作数据库，一个watch
在其中一个窗口执行如下命令，开启监听
2、在另一个窗口下，给上面的articledb插入一条数据
数据写入成功后，在第一个窗口下，执行下面的命令：
说明已经成功监听到新增的数据，修改、删除事件可以做类似的操作即可
以上先通过shell窗口展示了一下changestream的使用效果，接下来，将通过程序演示下如何在客户端集成并使用changestream
在未对articledb数据库下的comment集合做任何操作之前，由于watch为检测到任何数据变化，所以无法进入到while循环中，接下来，从shell端给comment集合新增一条数据，然后再次观察控制台数据变化
可以看到，控制台很快就检测到变化的数据
以下为完整的日志数据
{ 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的客户端操作类似，需要在一个配置类去监听即可
MongoMessageListener 类 ，顾名思义，该类用于监听特定数据库下的集合数据变化使用的，在实际开发中，该类的作用也是非常重要的，类似于许多中间件的客户端监听程序，当监听到数据变化后，做出后续的业务响应，比如，数据入库、推送消息到kafka、发送相关的事件等等
几乎是实时的监听到事件操作的数据变化，下面是完整的输出日志
测试2：通过shell窗口删除上面新增的这条数据
如果一个系统的数据需要迁移到另一个系统，可以考虑使用mongodb changestream这种方式，试想，如果老系统数据非常杂乱，并且文档中存在一些脏数据时，为了确保迁移后的数据能较快的投产，通过应用程序的方式，能够原始的数据做类似ETL的处理，这样更加方便
如果您的系统对数据监管较为严格，可以考虑使用changestream这种方式，订阅特定事件的数据操作，比如修改和删除数据的事件，然后及时的发送告警通知
我们知道，mongodb作为一款性能优秀的分布式文档型数据库，其实是可以存储海量数据的，在一些大数据场景下，比如下游其他的应用采用大数据技术，需要对mongo中的数据做轨迹行为分析，changestream就是一种不错的选择，当监听到特定事件的数据变化时，向消息队列，比如kafka推送相应的消息，下游相关的大数据应用就可以做后续的业务处理了
到此这篇关于springboot整合mongodb changestream的示例代码的文章就介绍到这了,更多相关springboot整合mongodb changestream内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！
