首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >MongoDB Change Stream初体验

MongoDB Change Stream初体验

作者头像
MongoDB中文社区
发布2019-07-08 12:17:12
8870
发布2019-07-08 12:17:12
举报
文章被收录于专栏:MongoDB中文社区MongoDB中文社区

Change Stream是MongoDB从3.6开始支持的新特性。这个新特性有哪些奇妙之处,会给我们带来什么便利?本次的文章将就这个主题进行初步讨论。

Change Stream 是什么 ?

顾名思义,Change Stream即变更流,是MongoDB向应用发布数据变更的一种方式。即当数据库中有任何数据发生变化,应用端都可以得到通知。我们可以将其理解为在应用中执行的触发器。至于应用想得到什么数据,以什么形式得到数据,则可以通过聚合框架加以过滤和转换。这点将在后文中讨论。

Change Stream 的原理

我们先来回顾一下MongoDB复制集大致是如何工作的:

  1. 应用通过驱动向数据库发起写入请求;
  2. 在同一个事务中,MongoDB完成oplog和集合的修改;
  3. oplog被其他从节点拉走;
  4. 从节点应用得到的oplog,同样在一个事务中完成对oplog和集合的修改;

至此,复制集同步完成。可以发现,整个同步过程是依赖于oplog来进行的。也就是说oplog实际上已经包含了我们需要的所有变更数据。如果观测oplog的变化,是否就能够得到所有变更的数据了呢?对,change stream正是基于这个原理实现的。但事情并没有这么简单!我们来看一下问题有可能出在什么地方。

如何从断点恢复

现实世界中,没有哪个应用是可以不间断运行的。不考虑bug导致的问题,正常的应用升级也会导致应用中断运行。那么在应用恢复的时候,从哪里开始继续获取变更呢?oplog当然是可以帮我们做到这点的,但你必须对MongoDB足够了解,才知道有oplogReplay这样的参数,以及其他一些问题。

如何有效地处理订阅

假设在一个应用中需要订阅10个不同集合的变更情况,是否需要开10个tailable cursor去获取oplog的变更呢?如果是100个集合呢?出于效率考虑显然不应该这么做。那么整个过程就会变成一个生产者-消费者模式,由一个线程负责从oplog获取变更,由订阅的线程负责消费这些变更。虽然实现也不是那么复杂,并且多半可以找到开源实现,但是涉及多线程就已经足够让初学者头疼一阵的了。 公平地说,上面这些还不算严重的问题,下面这些问题可能会更让人头疼。

如何管理权限

想要tail oplog,必须对local.oplog.rs有读权限。实际上这相当于对整个数据库都有了读权限,因为所有的变更都会在这里体现出来。DBA可能会阻止你这么做,因为这实在不是一个很安全的做法。

如何数据回滚

极端情况下,如果应用处理不当,MongoDB中可能发生数据回滚rollback的问题。如果仅仅通过跟踪oplog,则会出现已经通知出去的变更被回滚的情况。

幸运的是上面这些问题现在都不是问题了,因为change stream帮我们规避了这些复杂的细节。

使用方法

由于各种驱动都会有不同的语法和API,从shell中尝试使用change stream可能是最简便的方法。这并不妨碍你随后在各种驱动中的使用,因为shell中能实现的功能在驱动中一定有对应的语法。下面就以shell为例看看change stream应该如何使用。

打开一个shell,订阅你需要关注的集合 比如:

var cursor = db.bar.watch();

为了便于演示,我们在这个shell中不断遍历这个游标以获取新数据:

while(true) {
    if (cursor.hasNext()) {
        print(JSON.stringify(cursor.next()));
    }
}

打开另一个shell,向bar集合中插入一条数据:

db.bar.insert({y: 1})

此时第一个shell中会立即输出变更数据:

{"_id":{"_data":
{"$binary":"glzquiIAAAACRmRfaWQAZFzquiK0lDNo+K0DpwBaEARUMrm0ruVACoftuxjt1RtCBA==","$type":"00"}},
"operationType":"insert","fullDocument":{"_id":{"$oid":"5ceaba22b4943368f8ad03a7"},"y":1},"ns":
{"db":"test","coll":"bar"},"documentKey":{"_id":{"$oid":"5ceaba22b4943368f8ad03a7"}}}

这里的一些字段的简单介绍。更完整的介绍请查阅文档change events

  • _id: 用于恢复断点时使用。即知道这个值,应用断开后下次重启里就可以从这个断点之后开始恢复获得变更;
  • operationType: 操作类型,常见的值包括:
    • insert
    • update
    • delete
  • ns: 正在操作的命名空间
  • fullDocument: 完整的文档

从断点恢复

var cursor = db.bar.watch([], {resumeAfter: <\_id>})

此时使用hasNext()/next()即可获取到随后的变更。

注意事项

{readConcern: ‘majority’}

为了避免被回滚的更新被发布出去,change stream选择只在一个变更到达大多数节点(不可能被回滚)时,才会将这些变更发布到应用。使用的方式即{readConcern: "majority"}。因此以下这些情况下change stream都是不会向应用通知任何变更的:

  • 禁用了readConcern;
  • 从旧版本升级,但没有更新featureCompatibilityVersion;
  • PSA架构中S宕机;

断点可恢复时间

因为change stream是依赖于oplog工作的,自然也会面临oplog面临的所有问题。问题之一就是oplog被覆盖。因此想要保证断点可以恢复,必须保证应用在oplog window的时间内请求断点。

删除集合

如果在订阅集合变更过程中集合被删除,则会收到一条invalid信息通知,表示集合已不再可用:

{
    "_id" : {
        "_data" : BinData(0,"glzqxCcAAAACFFoQBFQyubSu5UAKh+27GO3VG0IE")
    },
    "operationType" : "invalidate"
}

参考资料

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-07-03,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Mongoing中文社区 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 如何从断点恢复
  • 如何有效地处理订阅
  • 如何管理权限
  • 如何数据回滚
  • {readConcern: ‘majority’}
  • 断点可恢复时间
  • 删除集合
相关产品与服务
云数据库 MongoDB
腾讯云数据库 MongoDB(TencentDB for MongoDB)是腾讯云基于全球广受欢迎的 MongoDB 打造的高性能 NoSQL 数据库,100%完全兼容 MongoDB 协议,支持跨文档事务,提供稳定丰富的监控管理,弹性可扩展、自动容灾,适用于文档型数据库场景,您无需自建灾备体系及控制管理系统。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档