Ouvrir un flux de changements sur une collection
Les apprenants appelleront watch() sur une collection et consommeront le flux d’événements dans une application Node.js avec une itération asynchrone.
Ouvrir un flux de changements sur une collection est une leçon MongoDB Academy gratuite sur CoddyKit. Ceci est la leçon 1 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage MongoDB Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours MongoDB Academy comprend 4 leçons au total.
Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.
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.
Questions Fréquemment Posées
La leçon « Ouvrir un flux de changements sur une collection » est-elle gratuite ?
Oui — le texte complet de « Ouvrir un flux de changements sur une collection » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours MongoDB Academy, passe à CoddyKit PRO. Le cours MongoDB Academy comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Ouvrir un flux de changements sur une collection » ?
Les apprenants appelleront watch() sur une collection et consommeront le flux d’événements dans une application Node.js avec une itération asynchrone. Tu pratiques MongoDB Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer MongoDB Academy ?
Aucune expérience préalable n'est requise. MongoDB Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 1 sur 4.
Combien de temps prend la leçon « Ouvrir un flux de changements sur une collection » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon MongoDB Academy ?
Oui. Chaque leçon MongoDB Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Ouvrir un flux de changements sur une collection
- Structure du document d’un événement de changement
- Filtrer des événements avec une chaîne de traitement d’agrégation
- Reprendre les flux de changements après une interruption