MongoDB Academy · 课时

在集合上打开变更流

您将对集合调用 watch()，并在 Node.js 应用中使用异步迭代消费事件流。

在集合上打开变更流 是 CoddyKit 上的免费 MongoDB Academy 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 MongoDB Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 MongoDB Academy 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

What Are Change Streams?

Change streams provide a real-time event feed of all insert, update, replace, delete, and invalidate operations on a MongoDB collection, database, or entire deployment. Introduced in MongoDB 3.6, they are built on top of the oplog (operation log)—the replica set's replication journal—but expose a high-level, resumable cursor API instead of requiring you to parse the raw oplog format.

Prerequisites for Change Streams

Change streams require a replica set or sharded cluster—they do not work on standalone MongoDB instances because they depend on the oplog. On MongoDB Atlas, all clusters (even the free tier M0) are replica sets, so change streams work out of the box. For local development, you need to start mongod with --replSet rs0 and initiate the replica set with rs.initiate() in mongosh.

// Verify you're on a replica set before using change streams
// In mongosh:
rs.status() // should show replica set members, not an error

// If running locally without a replica set, start one:
// mongod --replSet rs0 --port 27017 --dbpath /data/db
// Then in mongosh: rs.initiate()

Opening a Change Stream With watch()

Call collection.watch() to open a change stream on a specific collection. The method returns a ChangeStream cursor object that you can iterate using async iteration, the next() method, or event listeners. The stream stays open and delivers events as they occur. An empty watch() call with no pipeline receives all change event types on that collection.

const { MongoClient } = require('mongodb');
const client = new MongoClient(process.env.MONGODB_URI);

async function watchCollection() {
  const db = client.db('ecommerce');
  const orders = db.collection('orders');

  // Open a change stream on the orders collection
  const changeStream = orders.watch();

  console.log('Watching for changes...');
  // The stream is now open and ready to deliver events
  return changeStream;
}

Consuming Events With Async Iteration

The most readable way to consume change stream events in modern Node.js is with async iteration using for await...of. This syntax automatically handles the cursor's next() calls and pauses the loop between events. The loop runs indefinitely until the change stream is closed or an error occurs. Always wrap the loop in a try/finally to ensure the stream is closed when your process exits.

async function processChanges() {
  const changeStream = db.collection('orders').watch();
  try {
    for await (const change of changeStream) {
      console.log('Change received:', change.operationType);
      console.log('Document key:', change.documentKey._id);
      // Handle the change event
      await handleOrderChange(change);
    }
  } finally {
    await changeStream.close();
  }
}

Consuming Events With EventEmitter

Alternatively, you can use the change stream as a Node.js EventEmitter. This approach is familiar if you use streams elsewhere in your code. Register a 'change' event handler for normal events and an 'error' handler for connection issues. This style is useful when you want to react to events without blocking a function with an await loop.

const changeStream = db.collection('products').watch();

changeStream.on('change', (event) => {
  console.log('Product changed:', event.operationType, event.documentKey._id);
  if (event.operationType === 'update') {
    invalidateProductCache(event.documentKey._id);
  }
});

changeStream.on('error', (error) => {
  console.error('Change stream error:', error);
  // Implement reconnect logic or close gracefully
});

// Close when done
// changeStream.close();

Change Stream Scope: Collection, Database, Client

You can open change streams at three scopes: collection level (watches one collection), database level (watches all collections in a database), and client level (watches all databases and collections in the deployment). The broader the scope, the higher the event volume. Most applications watch specific collections to receive only the events they care about.

// Collection-level (most common)
const stream1 = db.collection('orders').watch();

// Database-level — all collections in 'ecommerce'
const stream2 = client.db('ecommerce').watch();

// Client-level — everything in the deployment
const stream3 = client.watch();

// Events at db/client scope include 'ns' field to identify which collection changed
for await (const change of stream2) {
  console.log('Changed collection:', change.ns.coll);
}

Change Stream Options: fullDocument

By default, update events only include the fields that changed (the update description), not the entire document. If you need the full updated document in the event payload, pass { fullDocument: 'updateLookup' } to watch(). This causes MongoDB to perform an additional lookup of the document after the update and include it in the change event. Be aware this adds latency and is a separate read after the event.

// Receive the full document in update events
const changeStream = db.collection('users').watch(
  [], // empty pipeline = all events
  { fullDocument: 'updateLookup' }
);

for await (const change of changeStream) {
  if (change.operationType === 'update') {
    // change.fullDocument is now the complete updated user document
    console.log('Updated user:', change.fullDocument.email);
    await syncToSearchIndex(change.fullDocument);
  }
}

Real-Time Dashboard Use Case

A classic use case for change streams is powering a live dashboard that shows new orders as they arrive. When a new order document is inserted, the change stream triggers, and your Node.js backend can push the update to connected clients via WebSocket or Server-Sent Events. This eliminates polling and provides true real-time updates without overloading the database with repeated queries.

// Server: push new orders to dashboard clients via WebSocket
async function startOrderWatcher(io) { // io = socket.io instance
  const changeStream = db.collection('orders').watch([
    { $match: { operationType: 'insert' } }
  ]);

  for await (const change of changeStream) {
    const newOrder = change.fullDocument;
    // Broadcast to all connected dashboard clients
    io.to('dashboard').emit('newOrder', {
      id: newOrder._id,
      customer: newOrder.customerId,
      amount: newOrder.total,
      timestamp: newOrder.createdAt
    });
  }
}

Event-Driven Microservices With Change Streams

Change streams can replace a message broker like Kafka or RabbitMQ for simple event-driven microservice patterns. When Service A writes to MongoDB, Service B watches the collection and reacts to changes. This avoids the operational complexity of a separate message bus for low-to-medium event volumes. However, for high-throughput use cases, dedicated message brokers offer better guarantees and higher throughput than change streams.

// Inventory service reacts to confirmed orders
async function inventoryWatcher() {
  const changeStream = db.collection('orders').watch([
    {
      $match: {
        operationType: 'update',
        'updateDescription.updatedFields.status': 'confirmed'
      }
    }
  ], { fullDocument: 'updateLookup' });

  for await (const change of changeStream) {
    const order = change.fullDocument;
    for (const item of order.items) {
      await reserveInventory(item.productId, item.quantity);
    }
  }
}

Closing a Change Stream Gracefully

Change streams consume a persistent connection to the MongoDB server. Always close them when your process shuts down or when you no longer need the stream. Call changeStream.close() which returns a Promise. For long-running applications, listen for process signals (SIGTERM, SIGINT) and close streams and clients before exiting to avoid resource leaks on the server side.

const changeStream = db.collection('events').watch();

// Graceful shutdown
process.on('SIGTERM', async () => {
  console.log('Shutting down...');
  await changeStream.close();
  await client.close();
  process.exit(0);
});

// Handle SIGINT (Ctrl+C in development)
process.on('SIGINT', async () => {
  await changeStream.close();
  await client.close();
  process.exit(0);
});

Change Streams vs Polling

Before change streams, applications relied on polling—periodically querying MongoDB to detect new or updated documents. Polling wastes resources (queries run even when nothing changed), introduces latency (the detection delay equals the poll interval), and scales poorly under high event rates. Change streams eliminate polling entirely: your application is notified immediately when data changes, consuming minimal resources when the collection is quiet. This makes change streams the preferred pattern for any feature that needs to react to data changes.

Quick Check

Test your understanding of MongoDB & NoSQL Databases concepts from this lesson.

Lesson Recap

In this lesson you learned: change streams provide a real-time event feed of all CRUD operations on a collection, database, or deployment, they require a replica set and are consumed via async iteration or EventEmitter, and the fullDocument option requests the complete updated document in update events. Next up we explore the structure of change event documents and how to handle each operation type.

常见问题解答

「在集合上打开变更流」课时是免费的吗？

是的 — 「在集合上打开变更流」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 MongoDB Academy 课程的其余内容，请升级到 CoddyKit PRO。 MongoDB Academy 课程共包含 4 节课。

「在集合上打开变更流」这节课中我会学到什么？

您将对集合调用 watch()，并在 Node.js 应用中使用异步迭代消费事件流。你通过在浏览器中直接运行的动手代码来练习 MongoDB Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 MongoDB Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 MongoDB Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「在集合上打开变更流」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 MongoDB Academy 课中编写并运行代码吗？

能。每节 MongoDB Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 MongoDB Academy