Monitor Data with Change Streams使用变更流监控数据

Iteration迭代

The following example opens a change stream on the haikus collection in the insertDB database and prints change events as they occur:以下示例在insertDB数据库中的haikus集合上打开一个更改流，并在更改事件发生时打印出来：

/* Change stream listener更改流侦听器 */

import { MongoClient } from "mongodb";

// Replace the uri string with your MongoDB deployment's connection string将uri字符串替换为MongoDB部署的连接字符串
const uri = "<connection string uri>";

const client = new MongoClient(uri);

const simulateAsyncPause = () =>
  new Promise(resolve => {
    setTimeout(() => resolve(), 1000);
  });

let changeStream;
async function run() {
  try {
    const database = client.db("insertDB");
    const haikus = database.collection("haikus");

    // Open a Change Stream on the "haikus" collection在“俳句”集合上打开更改流
    changeStream = haikus.watch();

    // Set up a change stream listener when change events are emitted在发出更改事件时设置更改流侦听器
    changeStream.on("change", next => {
      // Print any change event打印任何更改事件
      console.log("received a change to the collection: \t", next);
    });

    // Pause before inserting a document插入文档前暂停
    await simulateAsyncPause();

    // Insert a new document into the collection在集合中插入新文档
    await myColl.insertOne({
      title: "Record of a Shriveled Datum",
      content: "No bytes, no problem. Just insert a document, in MongoDB",
    });

    // Pause before closing the change stream关闭更改流前暂停
    await simulateAsyncPause();

    // Close the change stream and print a message to the console when it is closed关闭更改流，并在关闭时向控制台打印一条消息
    await changeStream.close();    
    console.log("closed the change stream");
  } finally {
    // Close the database connection on completion or error完成或出错时关闭数据库连接
    await client.close();
  }
}
run().catch(console.dir);

When you run this code and then make a change to the haikus collection, such as performing an insert or delete operation, you can see the change event document printed in your terminal.当您运行此代码并对haikus集合进行更改时，例如执行插入或删除操作，您可以在终端中看到打印的更改事件文档。

For example, if you insert a document to the collection, the code prints the following output:例如，如果将文档插入到集合中，代码将打印以下输出：

Received change:
{
  _id: {
    _data: '...'
  },
  operationType: 'insert',
  clusterTime: new Timestamp({ t: 1675800603, i: 31 }),
  fullDocument: {
    _id: new ObjectId("..."),
    ...
  },
  ns: { db: 'insertDB', coll: 'haikus' },
  documentKey: { _id: new ObjectId("...") }
}

const options = { fullDocument: "updateLookup" };
// This could be any pipeline.
const pipeline = [];

const changeStream = myColl.watch(pipeline, options);

Listener Function侦听器函数

The following example opens a change stream on the haikus collection in the insertDB database. Let's create a listener function to receive and print change events that occur on the collection.以下示例在insertDB数据库中的haikus集合上打开一个更改流。让我们创建一个监听器函数来接收和打印集合上发生的更改事件。

First, open the change stream on the collection and then define a listener on the change stream using the on() method. Once you set the listener, generate a change event by performing a change to the collection.首先，打开集合上的更改流，然后使用on()方法在更改流上定义一个监听器。设置侦听器后，通过对集合执行更改来生成更改事件。

To generate the change event on the collection, let's use the insertOne() method to add a new document. Since insertOne() may run before the listener function can register, we use a timer, defined as simulateAsyncPause to wait 1 second before executing the insert.要在集合上生成更改事件，让我们使用insertOne()方法添加一个新文档。由于insertOne()可能会在侦听器函数注册之前运行，因此我们使用一个计时器，定义为simulateAsyncPause，在执行插入之前等待1秒。

We also use simulateAsyncPause after the insertion of the document. 在插入文档后，我们还使用simulateAsyncPause。This provides ample time for the listener function to receive the change event and for the listener to complete its execution before closing the ChangeStream instance using the close() method.这为监听器函数接收更改事件以及监听器在使用close()方法关闭ChangeStream实例之前完成其执行提供了充足的时间。

/* Change stream listener更改流侦听器 */

import { MongoClient } from "mongodb";

// Replace the uri string with your MongoDB deployment's connection string将uri字符串替换为MongoDB部署的连接字符串
const uri = "<connection string uri>";

const client = new MongoClient(uri);

const simulateAsyncPause = () =>
  new Promise(resolve => {
    setTimeout(() => resolve(), 1000);
  });

let changeStream;
async function run() {
  try {
    const database = client.db("insertDB");
    const haikus = database.collection("haikus");

    // Open a Change Stream on the "haikus" collection在“俳句”集合上打开更改流
    changeStream = haikus.watch();

    // Set up a change stream listener when change events are emitted在发出更改事件时设置更改流侦听器
    changeStream.on("change", next => {
      // Print any change event打印任何更改事件
      console.log("received a change to the collection: \t", next);
    });

    // Pause before inserting a document插入文档前暂停
    await simulateAsyncPause();

    // Insert a new document into the collection在集合中插入新文档
    await myColl.insertOne({
      title: "Record of a Shriveled Datum",
      content: "No bytes, no problem. Just insert a document, in MongoDB",
    });

    // Pause before closing the change stream关闭更改流前暂停
    await simulateAsyncPause();

    // Close the change stream and print a message to the console when it is closed关闭更改流，并在关闭时向控制台打印一条消息
    await changeStream.close();    
    console.log("closed the change stream");
  } finally {
    // Close the database connection on completion or error完成或出错时关闭数据库连接
    await client.close();
  }
}
run().catch(console.dir);

Visit the following resources for more material on the classes and methods mentioned on this page:访问以下资源，了解更多关于本页提到的类和方法的材料：

• Change streams更改流
• Change events更改事件
• Aggregation pipeline聚合管道
• Aggregation stages聚合阶段
• ChangeStream class API documentationChangeStream类API文档
• Collection.watch(),
• Db.watch(),
• MongoClient.watch() API documentationAPI文档