Change Streams更改流

Availability可用性

Change streams are available for replica sets and sharded clusters:更改流可用于副本集和分片集群：

• Storage Engine.存储引擎。

  The replica sets and sharded clusters must use the WiredTiger storage engine. 副本集和分片集群必须使用WiredTiger存储引擎。Change streams can also be used on deployments that employ MongoDB's encryption-at-rest feature.变更流也可以用于使用MongoDB的静态加密功能的部署。

• Replica Set Protocol Version.副本集协议版本。

  The replica sets and sharded clusters must use replica set protocol version 1 (pv1).副本集和分片集群必须使用副本集协议版本1（pv1）。

• Read Concern "majority" Enablement.读取关注“多数”支持。

  Starting in MongoDB 4.2, change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabled to use change streams.从MongoDB 4.2开始，无论"majority"读取关注点支持如何，都可以使用更改流；也就是说，为了使用变更流，可以启用（默认）或禁用读取关注majority支持。

  In MongoDB 4.0 and earlier, change streams are available only if "majority" read concern support is enabled (default).在MongoDB 4.0及更早版本中，只有启用了“多数”读取关注点支持（默认），更改流才可用。

Connect连接

Connections for a change stream can either use DNS seed lists with the +srv connection option or by listing the servers individually in the connection string.更改流的连接可以使用带有+srv连接选项的DNS种子列表，也可以在连接字符串中单独列出服务器。

If the driver loses the connection to a change stream or the connection goes down, it attempts to reestablish a connection to the change stream through another node in the cluster that has a matching read preference. 如果驱动程序失去与更改流的连接或连接中断，它会尝试通过群集中具有匹配读取首选项的另一个节点重新建立与更改流之间的连接。If the driver cannot find a node with the correct read preference, it throws an exception.如果驱动程序找不到具有正确读取首选项的节点，则会引发异常。

For more information, see Connection String URI Format.有关详细信息，请参阅连接字符串URI格式。

Watch a Collection, Database, or Deployment监视集合、数据库或部署

You can open change streams against:您可以针对以下内容打开变更流：

Change Stream Performance Considerations变更流性能注意事项

If the amount of active change streams opened against a database exceeds the connection pool size, you may experience notification latency. 如果针对数据库打开的活动更改流的数量超过连接池大小，则可能会遇到通知延迟。Each change stream uses a connection and a getMore operation on the change stream for the period of time that it waits for the next event. 每个变更流在等待下一个事件的时间段内对变更流使用一个连接和getMore操作。To avoid any latency issues, you should ensure that the pool size is greater than the number of opened change streams. 为了避免任何延迟问题，您应该确保池的大小大于打开的更改流的数量。For details see the maxPoolSize setting.有关详细信息，请参阅maxPoolSize设置。

Open A Change Stream打开变更流

To open a change stream:要打开变更流：

• For a replica set, you can issue the open change stream operation from any of the data-bearing members.对于副本集，您可以从任何数据承载成员发出打开的更改流操作。
• For a sharded cluster, you must issue the open change stream operation from the mongos.对于分片集群，您必须从mongos发出开放的更改流操作。

The following example opens a change stream for a collection and iterates over the cursor to retrieve the change stream documents. 以下示例打开集合的变更流，并在游标上迭代以检索变更流文档。[1]

➤ Use the Select your language drop-down menu in the upper-right to set the language of the examples on this page.

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream.on('change', next => {
  // process next document
});

const collection = db.collection('inventory');
const changeStream = collection.watch();
const next = await changeStream.next();

To retrieve the data change event from the cursor, iterate the change stream cursor. 要从游标中检索数据更改事件，请迭代更改流游标。For information on the change stream event, see Change Events.有关更改流事件的信息，请参阅更改事件。

While the connection to the MongoDB deployment remains open, the cursor remains open until one of the following occurs:当与MongoDB部署的连接保持打开时，游标保持打开状态，直到出现以下情况之一：

• The cursor is explicitly closed.游标明确关闭。
• An invalidate event occurs.发生无效事件。
• If the deployment is a sharded cluster, a shard removal may cause an open change stream cursor to close, and the closed change stream cursor may not be fully resumable.如果部署是分片集群，则分片移除可能会导致打开的变更流游标关闭，并且关闭的变更流游标可能无法完全恢复。

Modify Change Stream Output修改更改流输出

➤ Use the Select your language drop-down menu in the upper-right to set the language of the examples on this page.

const pipeline = [
  { $match: { 'fullDocument.username': 'alice' } },
  { $addFields: { newField: 'this is an added field!' } }
];

const collection = db.collection('inventory');
const changeStream = collection.watch(pipeline);
changeStream.on('change', next => {
  // process next document
});

const changeStreamIterator = collection.watch(pipeline);
const next = await changeStreamIterator.next();

Lookup Full Document for Update Operations查找完整文档以进行更新操作

By default, change streams only return the delta of fields during the update operation. However, you can configure the change stream to return the most current majority-committed version of the updated document.默认情况下，更改流仅在更新操作期间返回字段的增量。但是，您可以将更改流配置为返回更新文档的最新多数提交版本。

➤ Use the Select your language drop-down menu in the upper-right to set the language of the examples on this page.

const collection = db.collection('inventory');
const changeStream = collection.watch([], { fullDocument: 'updateLookup' });
changeStream.on('change', next => {
  // process next document
});

const changeStreamIterator = collection.watch([], { fullDocument: 'updateLookup' });
const next = await changeStreamIterator.next();

Resume a Change Stream恢复更改流

Change streams are resumable by specifying a resume token to either resumeAfter or startAfter when opening the cursor.通过在打开游标时将恢复标记指定为resumeAfter或startAfter，可以恢复更改流。

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 })
}

Use Cases使用案例

Change streams can benefit architectures with reliant business systems, informing downstream systems once data changes are durable. 更改流可以使具有依赖业务系统的体系结构受益，一旦数据更改持久，就可以通知下游系统。For example, change streams can save time for developers when implementing Extract, Transform, and Load (ETL) services, cross-platform synchronization, collaboration functionality, and notification services.例如，在实现提取、转换和加载（ETL）服务、跨平台同步、协作功能和通知服务时，更改流可以为开发人员节省时间。

Access Control访问控制

For deployments enforcing Authentication and authorization:对于强制执行身份验证和授权的部署：

• To open a change stream against specific collection, applications must have privileges that grant changeStream and find actions on the corresponding collection.要针对特定集合打开更改流，应用程序必须具有授予changeStream并在相应集合上find操作的权限。

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

• To open a change stream on a single database, applications must have privileges that grant changeStream and find actions on all non-system collections in the database.要在单个数据库上打开更改流，应用程序必须具有授予changeStream的权限，并在数据库中的所有非system集合上find操作。

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

• To open a change stream on an entire deployment, applications must have privileges that grant changeStream and find actions on all non-system collections for all databases in the deployment.要在整个部署中打开更改流，应用程序必须具有授予changeStream的权限，并在部署中所有数据库的所有非system集合上find操作。

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

Event Notification事件通知

Change streams only notify on data changes that have persisted to a majority of data-bearing members in the replica set. 更改流仅通知已持久化到副本集中的大多数数据承载成员的数据更改。This ensures that notifications are triggered only by majority-committed changes that are durable in failure scenarios.这确保了通知仅由大多数提交的更改触发，这些更改在失败场景中是持久的。

For example, consider a 3-member replica set with a change stream cursor opened against the primary. 例如，考虑一个由3个成员组成的复制集，其中一个变更流游标对着primary打开。If a client issues an insert operation, the change stream only notifies the application of the data change once that insert has persisted to a majority of data-bearing members.如果客户端发出插入操作，则只有在插入持续存在于大多数数据承载成员之后，更改流才会向应用程序通知数据更改。

If an operation is associated with a transaction, the change event document includes the txnNumber and the lsid.如果操作与事务相关联，则更改事件文档包括txnNumber和lsid。

Collation排序规则

Starting in MongoDB 4.2, change streams use simple binary comparisons unless an explicit collation is provided. 从MongoDB 4.2开始，更改流使用simple二进制比较，除非提供明确的排序规则。In earlier versions, change streams opened on a single collection (db.collection.watch()) would inherit that collection's default collation.在早期版本中，在单个集合（db.collection.watch()）上打开的更改流将继承该集合的默认排序规则。

Change Streams and Orphan Documents更改流和孤立文档

Starting in MongoDB 5.3, during range migration, change stream events are not generated for updates to orphaned documents.从MongoDB 5.3开始，在范围迁移过程中，不会为孤立文档的更新生成更改流事件。

Change Streams with Document Pre- and Post-Images使用文档前映像和后映像更改流

Starting in MongoDB 6.0, you can use change stream events to output the version of a document before and after changes (the document pre- and post-images):从MongoDB 6.0开始，您可以使用更改流事件来输出更改前后文档的版本（文档前映像和后映像）：

• The pre-image is the document before it was replaced, updated, or deleted. 前映像是在文档被替换、更新或删除之前的文档。There is no pre-image for an inserted document.插入的文档没有前映像。
• The post-image is the document after it was inserted, replaced, or updated. 后映像是插入、替换或更新后的文档。There is no post-image for a deleted document.已删除的文档没有后映像。
• Enable changeStreamPreAndPostImages for a collection using db.createCollection(), create, or collMod.使用db.createCollection()、create或collMod为集合启用changeStreamPreAndPostImages。

Pre- and post-images are not available for a change stream event if the images were:如果映像为以下情况，则前映像和后映像对更改流事件不可用：

• Not enabled on the collection at the time of a document update or delete operation.文档更新或删除操作时未在集合上启用。
• Removed after the pre- and post-image retention time set in expireAfterSeconds.在expireAfterSeconds中设置的前映像和后映像保留时间之后删除。
  • The following example sets expireAfterSeconds to 100 seconds:以下示例将expireAfterSeconds设置为100秒：

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

  • The following example returns the current changeStreamOptions settings, including expireAfterSeconds:以下示例返回当前changeStreamOptions设置，包括expireAfterSeconds：

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

  • Setting expireAfterSeconds to off uses the default retention policy: pre- and post-images are retained until the corresponding change stream events are removed from the oplog.将expireAfterSeconds设置为off使用默认的保留策略：保留前映像和后映像，直到从oplog中删除相应的更改流事件。
  • If a change stream event is removed from the oplog, then the corresponding pre- and post-images are also deleted regardless of the expireAfterSeconds pre- and post-image retention time.如果更改流事件从操作日志中删除，则相应的前映像和后映像也将被删除，而不管映像前和后映像的保留时间是expireAfterSeconds。

Additional considerations:其他注意事项：

• Enabling pre- and post-images consumes storage space and adds processing time. 启用前映像和后映像会消耗存储空间并增加处理时间。Only enable pre- and post-images if you need them.只有在需要时才启用前映像和后映像。
• Limit the change stream event size to less than 16 megabytes. 将更改流事件大小限制为小于16兆字节。To limit the event size, you can:要限制事件大小，您可以：
  • Limit the document size to 8 megabytes. 将文档大小限制为8兆字节。You can request pre- and post-images simultaneously in the change stream output if other change stream event fields like updateDescription are not large.如果其他变更流事件字段（如updateDescription）不大，则可以在变更流输出中同时请求前映像和后映像。
  • Request only post-images in the change stream output for documents up to 16 megabytes if other change stream event fields like updateDescription are not large.如果其他更改流事件字段（如updateDescription）不太大，则请求在更改流输出中为高达16兆字节的文档后映像。
  • Request only pre-images in the change stream output for documents up to 16 megabytes if:如果出现以下情况，则仅在更改流输出中请求高达16兆字节的文档的前映像：
    • document updates affect only a small fraction of the document structure or content, and文档更新只影响文档结构或内容的一小部分，并且
    • do not cause a replace change event. A replace event always includes the post-image.不要引起replace更改事件。replace事件始终包括后映像。
• To request a pre-image, you set fullDocumentBeforeChange to required or whenAvailable in db.collection.watch(). 若要请求前映像，请在db.collection.watch()中将fullDocumentBeforeChange设置为required或whenAvailable。To request a post-image, you set fullDocument using the same method.若要请求后映像，请使用相同的方法设置fullDocument。
• Pre-images are written to the config.system.preimages collection.前映像将写入config.system.preimages集合。
  • The config.system.preimages collection may become large. config.system.preimages集合可能会变大。To limit the collection size, you can set expireAfterSeconds time for the pre-images as shown earlier.要限制集合大小，您可以为前映像设置expireAfterSeconds时间，如前所示。
  • Pre-images are removed asynchronously by a background process.前映像通过后台进程异步移除。

For complete examples with the change stream output, see Change Streams with Document Pre- and Post-Images.有关更改流输出的完整示例，请参阅使用文档前映像和后映像的更改流。