MongoDB 变更流

可用性

变更流可用于副本集和分片集群：

• 存储引擎。

  副本集和分片集群必须使用 WiredTiger 存储引擎。变更流也可以用于使用静态加密功能的部署。

• 副本集协议版本。

  副本集和分片集群必须使用副本集协议版本 1 (pv1)。

• 读关注“多数”启用。

  从 MongoDB 4.2 开始，无论是否有 "majority" 读关注支持，change stream 都可用；也就是说，读关注 majority 支持可以启用（默认），也可以禁用以使用 change stream。

  在 MongoDB 4.0 及更早版本中，仅当启用 "majority" 读关注支持时，才能使用变更流。

连接

连接 change stream 可以使用带有+srv连接选项的 DNS 种子列表，也可以在连接字符串中单独列出服务器。

如果驱动程序与变更流失去连接或连接中断，它则会尝试通过集群中具有匹配读取偏好的其他节点与变更流重新建立连接。如果驱动程序未找到具有正确读取偏好的节点，则会引发异常。

有关更多信息，请参阅连接字符串 URI 格式。

监视集合、数据库或部署

可以针对如下情况打开变更流：

变更流性能考量

如果针对数据库打开的活动变更流的数量超过连接池大小，则可能会出现通知延迟。每个变更流在等待下一个事件时，都会通过 getMore 操作保持一个连接处于打开状态。为避免产生延迟，请确保池大小应大于已打开的变更流数量。有关详情，请参阅 maxPoolSize 设置。

打开变更流

要打开变更流：

• 对于副本集，请从任何承载数据的节点打开变更流。

• 对于分片集群，从 mongos 打开变更流。

The following example opens a change stream for a collection and iterates over the cursor to retrieve the change stream documents. [1]

迭代变更流游标以检索变更事件。有关变更流文档格式，请参阅“变更流事件”。

变更流游标保持打开状态，直到出现以下任一情况：

• 游标已明确关闭。

• 发生失效事件；例如删除或重命名集合。

• 与MongoDB 部署的连接关闭或超时。有关更多信息，请参阅行为。

• 如果部署是分片集群，则分片删除可能会导致打开的变更流游标关闭，而关闭的变更流游标可能无法完全恢复。

修改变更流输出

查找更新操作的完整文档

默认情况下，变更流仅在更新操作期间返回字段的增量。不过，您可以配置变更流以返回已更新文档的最新多数提交版本。

updateLookup 操作从集合中读取由分片键和文档标识符标识的文档。集合由名称标识，并使用处理变更流时存在的集合数据。考虑以下场景：

• 如果重命名了集合，则不会返回任何文档。

• 如果重命名了集合，并使用旧名称创建了新集合，则会对新集合执行查找操作。如果找到匹配的文档，则将其返回。

使用 documentKey 作为查询筛选器

您可以使用变更事件中的 documentKey字段来查询该事件描述的文档。要按字面意思匹配每个字段值，请将 documentKey 中的每个字段值包装在 $eq操作符中。

以下示例在 orders集合上打开一个变更流。对于每个事件，该示例从 documentKey 构建一个过滤，并使用它在 ordersArchive集合中查找匹配文档：

const changeStream = db.orders.watch();
while (changeStream.hasNext()) {
   const event = changeStream.next();
   // Skip DDL events without a documentKey
   if (!event.documentKey) {
      continue;
   }
   // Wrap each documentKey field value in $eq
   const filter = {};
   for (const [field, value] of Object.entries(event.documentKey)) {
      filter[field] = { $eq: value };
   }
   printjson(db.ordersArchive.find(filter).toArray());
}

对于具有以下 documentKey 的事件：

documentKey: { _id: 1, region: "west" }

该示例构建此过滤：

{ _id: { $eq: 1 }, region: { $eq: "west" } }

// Shard key: { region: 1, _id: 1 }
// documentKey from the event: { _id: 1 }
{ region: { $eq: null }, _id: { $eq: 1 } }

恢复变更流

Change streams are resumable by specifying a resume token to either resumeAfter or startAfter when opening the cursor.

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

npx mongodb-resumetoken-decoder <RESUME TOKEN>

{
   "_id": {
      "_data": "82635019A0000000012B042C0100296E5A1004AB1154ACACD849A48C61756D70D3B21F463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F69640064635019A078BE67426D7CF4D2000004"
    },
    "operationType": "insert",
    "clusterTime": Timestamp({ "t": 1666193824, "i": 1 }),
    "collectionUUID": new UUID("ab1154ac-acd8-49a4-8c61-756d70d3b21f"),
    "wallTime": ISODate("2022-10-19T15:37:04.604Z"),
    "fullDocument": {
       "_id": ObjectId("635019a078be67426d7cf4d2"'),
       "name": "Giovanni Verga"
    },
    "ns": {
       "db": "test",
       "coll": "names"
    },
    "documentKey": {
       "_id": ObjectId("635019a078be67426d7cf4d2")
    }
}

{
   "cursor": {
      "firstBatch": [],
      "postBatchResumeToken": {
         "_data": "8263515EAC000000022B0429296E1404"
      },
      "id": Long("4309380460777152828"),
      "ns": "test.names"
   },
   "ok": 1,
   "$clusterTime": {
      "clusterTime": Timestamp({ "t": 1666277036, "i": 1 }),
      "signature": {
         "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
         "keyId": Long("0")
      }
   },
   "operationTime": Timestamp({ "t": 1666277036, "i": 1 })
}

{
   "cursor": {
      "nextBatch": [],
      "postBatchResumeToken": {
         "_data": "8263515979000000022B0429296E1404"
      },
      "id": Long("7049907285270685005"),
      "ns": "test.names"
   },
   "ok": 1,
   "$clusterTime": {
      "clusterTime": Timestamp( { "t": 1666275705, "i": 1 } ),
      "signature": {
         "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
         "keyId": Long("0")
      }
   },
   "operationTime": Timestamp({ "t": 1666275705, "i": 1 })
}

用例

变更流对于采用业务依赖型系统的基础设施很有益处，因为数据更改一旦变为持久更改，它就会通知下游系统。例如，在实施提取、转换和加载 (ETL) 服务、跨平台同步、协作功能以及通知服务时，变更流可为开发人员节省时间。

访问控制

对于在自管理部署上执行身份验证和授权的部署：

• 要打开针对特定集合的变更流，应用程序必须具有对相应集合授予 changeStream 和 find 动作的特权。

  { resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }

• 要在单个数据库上打开变更流，应用程序必须具有对数据库中所有非 system 集合授予 changeStream 和 find 动作的特权。

  { resource: { db: <dbname>, collection: "" }, actions: [ "find", "changeStream" ] }

• 要在整个部署中打开变更流，应用程序必须具有对部署中所有数据库的所有非 system 集合授予 changeStream 和 find 动作的特权。

  { resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

事件通知

变更流仅在数据发生更改时通知副本集中的大多数数据承载节点。变更流确保通知仅由大多数已提交且在故障情况下持续存在的更改触发。

例如，考虑一个 3 节点副本集，针对主节点打开了变更流游标。如果客户端发出插入操作，则只有在插入持续到大多数数据承载节点后，变更流才会将数据更改通知应用程序。

如果某个操作与事务相关联，则变更事件文档包括 txnNumber 和 lsid。

排序规则

除非提供了显式排序规则，否则变更流使用 simple 二进制比较。

变更流和孤立文档

从 MongoDB 5.3 开始，在范围迁移期间，不会为孤立文档的更新生成 change stream 事件。

附带文档前映像和后映像的变更流

从 MongoDB 6.0 开始，可使用变更流事件来输出更改前后的文档版本（文档前映像和后映像）：

• 前映像是指被替换、更新或删除之前的文档。已插入的文档没有前映像。

• 后图像是插入、替换或更新后的文档。 已删除的文档没有后图像。

• 使用 db.createCollection()、create 或 collMod 为集合启用 changeStreamPreAndPostImages。示例，使用 collMod 命令时：

  db.runCommand( {
     collMod: <collection>,
     changeStreamPreAndPostImages: { enabled: true }
  } )

要检查集合的当前 changeStreamPreAndPostImages 设置，运行 db.getCollectionInfos():

db.getCollectionInfos( {
  name: "<collection>"
} )[0].options.changeStreamPreAndPostImages

{ changeStreamPreAndPostImages: { enabled: true } }

如果图像属于以下情况，则前像和后像不可用于变更流事件：

• 在文档更新或删除操作时未对集合启用。

• 在 expireAfterSeconds 中设置的前像和后像保留时间后之后被删除。

  • 以下示例将整个集群上的 expireAfterSeconds 设置为 100 秒：

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: {
             preAndPostImages: { expireAfterSeconds: 100 }
          } }
    } )

    注意

    The setClusterParameter command is not supported in MongoDB Atlas clusters. For information on Atlas support for all commands, see Unsupported Commands in Atlas Clusters.

  • 以下示例返回当前的 changeStreamOptions 设置，包括 expireAfterSeconds：

    db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

  • 将 expireAfterSeconds 设置为 off 可使用默认保留策略：将保留前像和后像，直到从 oplog 中删除对应的变更流事件。

  • 如果变更流事件从 oplog 中删除，则无论 expireAfterSeconds 前映像和后映像保留时间如何，相应的前映像和后映像也会被删除。

其他考量：

• 启用前像和后像会占用存储空间并增加处理时间。仅在需要时启用前像和后像。

• 将变更流事件大小限制为小于 16 MiB。要限制事件大小，您可以：

  • 将文档大小限制为 8 MB。如果其他 change stream 事件字段（例如 updateDescription）不是很大，则可以在 change stream 输出中同时请求更新前的文档和更新后的文档。

  • 如果其他变更流事件字段（例如 updateDescription）并不大，则仅请求变更流输出中最多 16 MiB 的文档的后像。

  • 在以下情况下，仅请求变更流输出中最多 16 MiB 的文档的前像：

    • 文档更新仅影响文档结构或内容的一小部分，且

    • 不会引起 replace 变更事件。replace 事件始终包含后像。

• 要请求前图像，请在db.collection.watch()中将fullDocumentBeforeChange设置为required或whenAvailable。要请求后图像，您可以使用相同的方法设置fullDocument。

• 前像被写入 config.system.preimages 集合。

  • config.system.preimages 集合可能会变大。要限制集合大小，可如前文所示为前映像设置 expireAfterSeconds 时间。

  • 要监控 config.system.preimages 的大小，请连接到分片集群上的分片节点或副本集上的 mongod 节点。然后，运行以下命令：

    use config
    db.system.preimages.totalSize()
    db.system.preimages.stats()

    注意

    您必须对 config.system.preimages 集合拥有 collStats 权限操作才能运行这些命令。

    要在 MongoDB Atlas 部署上运行这些命令，您必须具有 atlasAdmin 角色。

  • 前像由后台进程异步删除。

有关变更流输出的完整示例，请参阅使用文档前像和后像的变更流。