第二十天

这几天有项目准备上线，所以学习的进度有点受影响。今天要习的是《23 | Change Stream》章节，主要讲解Change Stream，有点类似于关系数据库的触发器。

什么是 Change Stream

Change Stream 是 MongoDB 用于实现变更追踪的解决方案，类似于关系数据库的触发器，但原理不完全相同。

Change Stream 的实现原理

Change Stream 是基于 oplog 实现的。它在 oplog 上开启一个 tailable cursor 来追踪所有复制集上的变更操作，最终调用应用中定义的回调函数。被追踪的变更事件主要包括：• insert/update/delete：插入、更新、删除；• drop：集合被删除；• rename：集合被重命名；• dropDatabase：数据库被删除；• invalidate：drop/rename/dropDatabase 将导致 invalidate 被触发，并关闭 change stream；

Change Stream 与可重复读

Change Stream 只推送已经在大多数节点上提交的变更操作。即“可重复读”的变更。这个验证是通过 {readConcern: “majority”} 实现的。因此：• 未开启 majority readConcern 的集群无法使用 Change Stream；• 当集群无法满足 {w: “majority”} 时，不会触发 Change Stream(例如 PSA 架构中的 S 因故障宕机)。

Change Stream 变更过滤

如果只对某些类型的变更事件感兴趣，可以使用使用聚合管道的过滤步骤过滤事件。例如：

var cs = db.collection.watch([{

$match: {

operationType: {

$in: ['insert', 'delete']

}

}

}])

Change Stream 示例

仍在集群环境下测试主节点可始监视

> db.collection.watch([],{maxAwaitTimeMS: 30000}).pretty()

从结点输入 insert 内容

> db.collection.insert({

_id: 1,

text: "hello"

})

主结点的反应：

{

_id : (resumeToken),

operationType : "insert",

...

fullDocument : {

_id : 1,

text: "hello"

}

}

Change Stream 故障恢复

假设在一系列写入操作的过程中，订阅 Change Stream 的应用在接收到“写3”之后于 t0 时刻崩溃，重启后后续的变更怎么办？

Change Stream 故障恢复

想要从上次中断的地方继续获取变更流，只需要保留上次变更通知中的 _id 即可。

每条这样的 Change Stream 返回数据都带有一个 _id，这个 _id 可以用于断点恢复。

例如：

var cs = db.collection.watch([], {resumeAfter: <_id>})

即可从上一条通知中断处继续获取后续的变更通知。

Change Stream 使用场景

• 跨集群的变更复制——在源集群中订阅 Change Stream，一旦得到任何变更立即写入目标集群。• 微服务联动——当一个微服务变更数据库时，其他微服务得到通知并做出相应的变更。• 其他任何需要系统联动的场景

注意事项

• Change Stream 依赖于 oplog，因此中断时间不可超过 oplog 回收的最大时间窗；• 在执行 update 操作时，如果只更新了部分数据，那么 Change Stream 通知的也是增量部分；• 同理，删除数据时通知的仅是删除数据的 _id。

作者：xiaopohair

原文地址：https://segmentfault.com/a/1190000021529129

喜欢 0

参考链接

评论可见，请评论后查看内容，谢谢！！！评论后请刷新页面。