Watch for Changes留意变化¶

Open a Change Stream打开变更流¶

You can watch for changes to a single collection, a database, or an entire deployment in MongoDB with Change Streams.您可以使用变更流监视MongoDB中单个集合、数据库或整个部署的变更。Open a change stream by calling the watch() method on a Collection, Db, or MongoClient object.通过对Collection、Db或MongoClient对象调用watch()方法打开更改流。The change stream emits change event documents when they occur.变更流在发生变更事件文档时发出变更事件文档。

The watch() method optionally takes an aggregation pipeline which consists of an array of aggregation stages as the first parameter.watch()方法可以选择将包含聚合阶段数组的聚合管道作为第一个参数。The aggregation stages filter and transform the change events.聚合阶段过滤和转换更改事件。

In the following snippet, the $match stage matches all change event documents with a runtime value of less than 15, filtering all others out.在下面的代码段中，$match阶段匹配runtime值小于15的所有更改事件文档，过滤掉所有其他文档。

const pipeline = [ { $match: { runtime: { $lt: 15 } } } ];
const changeStream = collection.watch(pipeline);

The watch() method accepts an options object as the second parameter.watch()方法接受options对象作为第二个参数。Refer to the links at the end of this section for more information on the settings you can configure with this object.有关可使用此对象配置的设置的更多信息，请参阅本节末尾的链接。

The watch() method returns an instance of a ChangeStream.watch()方法返回ChangeStream的实例。You can read events from change streams by iterating over them or listening for events.您可以通过迭代更改流或侦听事件，从更改流中读取事件。Select the tab that corresponds to the way you want to read events from the change stream below.选择与要从下面的更改流中读取事件的方式相对应的选项卡。

changeStream.on("change", (changeEvent) => { /* your callback function */ });

changeStream.close();

Visit the following resources for additional material on the classes and methods presented above:有关上述类和方法的更多资料，请访问以下资源：

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

Example示例¶

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

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

To generate the change event on the collection, let's use insertOne() method to add a new document. 要在集合上生成更改事件，让我们使用insertOne()方法添加一个新文档。Since the 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()可能在侦听器函数注册之前运行，因此我们使用一个定义为simulateAsyncPause的计时器在执行insert之前等待1秒。

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

The timers used in this example are only necessary for this demonstration to make sure there is enough time to register listener and have the callback process the event before exiting.本示例中使用的计时器仅在本演示中是必要的，以确保有足够的时间注册侦听器，并在退出之前让回调处理事件。

import { MongoClient } from "mongodb";

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

const client = new MongoClient(uri);

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

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

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

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

    await simulateAsyncPause();

    await collection.insertOne({
      title: "Record of a Shriveled Datum",
      content: "No bytes, no problem. Just insert a document, in MongoDB",
    });

    await simulateAsyncPause();

    await changeStream.close();
    console.log("closed the change stream");
  } finally {
    await client.close();
  }
}
run().catch(console.dir);

import { MongoClient } from "mongodb";


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

const client = new MongoClient(uri);

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

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

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

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

    await simulateAsyncPause();

    await collection.insertOne({
      title: "Record of a Shriveled Datum",
      content: "No bytes, no problem. Just insert a document, in MongoDB",
    });

    await simulateAsyncPause();

    await changeStream.close();
    console.log("closed the change stream");
  } finally {
    await client.close();
  }
}
run().catch(console.dir);

If you run the preceding example, you should see the following output:如果运行前面的示例，您应该会看到以下输出：

received a change to the collection:          {
  _id: { _data: '825EC...' },
  operationType: 'insert',
  clusterTime: new Timestamp { ... },
  fullDocument: { _id: new ObjectId(...), title: 'Record of a Shriveled Datum', content: 'No bytes, no problem. Just insert a document, in MongoDB' },
  ns: { db: 'insertDB', coll: 'haikus' },
  documentKey: { _id: new ObjectId(...) }
}
closed the change stream

const options = { fullDocument: "updateLookup" };
// This could be any pipeline.这可能是任何管道。
const pipeline = [];

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