怎么做点击图片跳转网站,前几年做那个网站致富,wordpress智能小程序,九龙坡网站建设哪家好MongoDB 从3.6版本开始支持了 Change Stream 能力#xff08;4.0、4.2 版本在能力上做了很多增强#xff09;#xff0c;用于订阅 MongoDB 内部的修改操作#xff0c;change stream 可用于 MongoDB 之间的增量数据迁移、同步#xff0c;也可以将 MongoDB 的增量订阅应用到…MongoDB 从3.6版本开始支持了 Change Stream 能力4.0、4.2 版本在能力上做了很多增强用于订阅 MongoDB 内部的修改操作change stream 可用于 MongoDB 之间的增量数据迁移、同步也可以将 MongoDB 的增量订阅应用到其他的关联系统比如电商场景里MongoDB 里存储新的订单信息业务需要根据新增的订单信息去通知库存管理系统发货。
Change Stream 与 Tailing Oplog 对比
在 change stream 功能之前如果要获取 MongoDB 增量的修改可以通过不断 tailing oplog 的方式来 拉取增量的 oplog 然后针对拉取到的 oplog 集合来过滤满足条件的 oplog。这种方式也能满足绝大部分场景的需求但存在如下的不足。
使用门槛较高用户需要针对 oplog 集合打开特殊选项的的 tailable cursor (tailable: true, awaitData : true)。用户需要自己管理增量续传当拉取应用 crash 时用户需要记录上一条拉取oplog的 ts、h 等字段在下一次先定位到指定 oplog 再继续拉取。结果过滤必须在拉取侧完成但只需要订阅部分 oplog 时比如针对某个 DB、某个 Collection、或某种类型的操作必须要把左右的 oplog 拉取到再进行过滤。对于 update 操作oplog 只包含操作的部分内容比如 {$set: {x: 1}} 而应用经常需要获取到完整的文档内容。不支持 Sharded Cluster 的订阅用户必须针对每个 shard 进行 tailing oplog并且这个过程中不能有 moveChunk 操作否则结果可能乱序。
MongoDB Change Stream 解决了 Tailing oplog 存在的不足
简单易用提供统一的 Change Stream API一次 API 调用即可从 MongoDB Server 侧获取增量修改。统一的进度管理通过 resume token 来标识拉取位置只需在 API 调用时带上上次结果的 resume token即可从上次的位置接着订阅。支持对结果在 Server 端进行 pipeline 过滤减少网络传输支持针对 DB、Collection、OperationType 等维度进行结果过滤。支持 fullDocument: updateLookup 选项对于 update返回当时对应文档的完整内容。支持 Sharded Cluster 的修改订阅相同的 API 请求发到 mongos 即可获取集群维度全局有序的修改。
Change Stream 实战
以 Mongo shell 为例使用 Change Stream 非常简单mongo shell 封装了针对整个实例、DB、Collection 级别的订阅操作。
db.getMongo().watch() 订阅整个实例的修改
db.watch() 订阅指定DB的修改
db.collection.watch() 订阅指定Collection的修改
新建连接1发起订阅操作
mytest:PRIMARYdb.coll.watch([], {maxAwaitTimeMS: 60000}) 最多阻塞等待 1分钟
新建连接2写入新数据mytest:PRIMARY db.coll.insert({x: 100})
WriteResult({ nInserted : 1 })
mytest:PRIMARY db.coll.insert({x: 101})
WriteResult({ nInserted : 1 })
mytest:PRIMARY db.coll.insert({x: 102})
WriteResult({ nInserted : 1 })
连接1上收到 Change Stream 更新
mytest:PRIMARY db.watch([], {maxAwaitTimeMS: 60000})
{ _id : { _data : 825E0D5E35000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E353BE5C36D695042C90004 }, operationType : insert, clusterTime : Timestamp(1577934389, 1), fullDocument : { _id : ObjectId(5e0d5e353be5c36d695042c9), x : 100 }, ns : { db : test, coll : coll }, documentKey : { _id : ObjectId(5e0d5e353be5c36d695042c9) } }
{ _id : { _data : 825E0D5E37000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E373BE5C36D695042CA0004 }, operationType : insert, clusterTime : Timestamp(1577934391, 1), fullDocument : { _id : ObjectId(5e0d5e373be5c36d695042ca), x : 101 }, ns : { db : test, coll : coll }, documentKey : { _id : ObjectId(5e0d5e373be5c36d695042ca) } }
{ _id : { _data : 825E0D5E39000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E393BE5C36D695042CB0004 }, operationType : insert, clusterTime : Timestamp(1577934393, 1), fullDocument : { _id : ObjectId(5e0d5e393be5c36d695042cb), x : 102 }, ns : { db : test, coll : coll }, documentKey : { _id : ObjectId(5e0d5e393be5c36d695042cb) } } 上述 ChangeStream 结果里_id 字段的内容即为 resume token标识着 oplog 的某个位置如果想从某个位置继续订阅在 watch 时通过 resumeAfter 指定即可。比如每个应用订阅了上述3条修改但只有第一条已经成功消费了下次订阅时指定第一条的 resume token 即可再次订阅到接下来的2条。mytest:PRIMARY db.coll.watch([], {maxAwaitTimeMS: 60000, resumeAfter: { _data : 825E0D5E35000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E353BE5C36D695042C90004 }})
{ _id : { _data : 825E0D5E37000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E373BE5C36D695042CA0004 }, operationType : insert, clusterTime : Timestamp(1577934391, 1), fullDocument : { _id : ObjectId(5e0d5e373be5c36d695042ca), x : 101 }, ns : { db : test, coll : coll }, documentKey : { _id : ObjectId(5e0d5e373be5c36d695042ca) } }
{ _id : { _data : 825E0D5E39000000012B022C0100296E5A1004EA4E00977BCC482FB44DEED9A3C2999946645F696400645E0D5E393BE5C36D695042CB0004 }, operationType : insert, clusterTime : Timestamp(1577934393, 1), fullDocument : { _id : ObjectId(5e0d5e393be5c36d695042cb), x : 102 }, ns : { db : test, coll : coll }, documentKey : { _id : ObjectId(5e0d5e393be5c36d695042cb) } }
Change Stream 内部实现
watch() wrapper
db.watch() 实际上是一个 API wrapper实际上 Change Stream 在 MongoDB 内部实际上是一个 aggregation 命令只是加了一个特殊的 $changestream 阶段在发起 change stream 订阅操作后可通过 db.currentOp() 看到对应的 aggregation/getMore 操作的详细参数。
{op : getmore,ns : test.coll,command : {getMore : NumberLong(233479991942333714),collection : coll,maxTimeMS : 50000,lsid : {id : UUID(e4fffa71-e168-4527-be61-f0918849d107)},},planSummary : COLLSCAN,cursor : {cursorId : NumberLong(233479991942333714),createdDate : ISODate(2019-12-31T06:35:52.479Z),lastAccessDate : ISODate(2019-12-31T06:36:09.988Z),nDocsReturned : NumberLong(1),nBatchesReturned : NumberLong(1),noCursorTimeout : false,tailable : true,awaitData : true,originatingCommand : {aggregate : coll,pipeline : [{$changeStream : {fullDocument : default}}],cursor : {},lsid : {id : UUID(e4fffa71-e168-4527-be61-f0918849d107)},$clusterTime : {clusterTime : Timestamp(1577774144, 1),signature : {hash : BinData(0,AAAAAAAAAAAAAAAAAAAAAAAAAAA),keyId : NumberLong(0)}},$db : test},operationUsingCursorId : NumberLong(7019500)},numYields : 2,locks : {}}
resume token
resume token 用来描述一个订阅点本质上是 oplog 信息的一个封装包含 clusterTime、uuid、documentKey等信息当订阅 API 带上 resume token 时MongoDB Server 会将 token 转换为对应的信息并定位到 oplog 起点继续订阅操作。
struct ResumeTokenData {Timestamp clusterTime;int version 0;size_t applyOpsIndex 0;Value documentKey;boost::optionalUUID uuid;
};
ResumeTokenData 结构里包含 version 信息在 4.0.7 以前的版本version 均为0 4.0.7 引入了一种新的 resume token 格式version 为 1 另外在 3.6 版本里Resume Token 的编码与 4.0 也有所不同所以在版本升级后有可能出现不同版本 token 无法识别的问题所以尽量要让 MongoDB Server 所有组件Replica Set 各个成员ConfigServer、Mongos都保持相同的内核版本。
updateLookup
Change Stream 支持针对 update 操作获取当前的文档完整内容而不是仅更新操作本身比如
mytest:PRIMARY db.coll.find({_id: 101})
{ _id : 101, name : jack, age : 18 }
mytest:PRIMARY db.coll.update({_id: 101}, {$set: {age: 20}})
WriteResult({ nMatched : 1, nUpserted : 0, nModified : 1 })
上面的 update 操作默认情况下change stream 会收到 {_id: 101}, {$set: {age: 20} 的内容而并不会包含这个文档其他未更新字段的信息而加上 fullDocument: updateLookup 选项后Change Stream 会根据文档 _id 去查找文档当前的内容并返回。
需要注意的是updateLookup 选项只能保证最终一致性比如针对上述文档如果连续更新100次update 的 change stream 并不会按顺序收到中间每一次的更新因为每次都是去查找文档当前的内容而当前的内容可能已经被后续的修改覆盖。
Sharded cluster
Change Stream 支持针对 sharded cluster 进行订阅会保证全局有序的返回结果为了达到全局有序这个目标mongos 需要从每个 shard 都返回订阅结果按时间戳进行排序合并返回。
在极端情况下如果某些 shard 写入量很少或者没有写入change stream 的返回延时会受到影响因为需要等到所有 shard 都返回订阅结果默认情况下mongod server 每10s会产生一条 Noop 的特殊oplog这个机制会间接驱动 sharded cluster 在写入量不高的情况下也能持续运转下去。
由于需要全局排序在 sharded cluster 写入量很高时Change Stream 的性能很可能跟不上如果对性能要求非常高可以考虑关闭 Balancer在每个 shard 上各自建立 Change Stream。
原文链接 本文为阿里云原创内容未经允许不得转载。
