Åbning af en ændringsstrøm på en samling
De lærende vil kalde watch() på en samling og modtage hændelsesstrømmen i en Node.js-applikation ved hjælp af asynkron iteration.
Åbning af en ændringsstrøm på en samling er en gratis MongoDB Academy-lektion på CoddyKit. Dette er lektion 1 af 4. Du kan læse hele lektionen gratis nedenfor — og derefter øve dig praktisk i browseren med en indbygget kodeeditor og en AI-vejleder, der er tilgængelig døgnet rundt. Den er en del af læringsforløbet i MongoDB Academy, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. MongoDB Academy-kurset indeholder 4 lektioner i alt.
Hvad er change streams?
Change streams leverer et realtidsbaseret hændelsesfeed med alle insert-, update-, replace-, delete- og invalidate-operationer på en MongoDB-samling, database eller hel installation. De blev introduceret i MongoDB 3.6 og bygger oven på oploggen (operationsloggen) – replica settets replikeringsjournal – men stiller en abstrakt cursor-API, der kan genoptages, til rådighed i stedet for at kræve, at du fortolker det rå oplog-format.
Forudsætninger for change streams
Change streams kræver et replica set eller sharded cluster – de fungerer ikke på selvstændige MongoDB-instanser, fordi de afhænger af oploggen. I MongoDB Atlas er alle clusters (selv det gratis M0-niveau) replica sets, så change streams fungerer med det samme. Til lokal udvikling skal du starte mongod med --replSet rs0 og initialisere replica settet med rs.initiate() i 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()Åbn et change stream med watch()
Kald collection.watch() for at åbne et change stream på en bestemt samling. Metoden returnerer et ChangeStream-cursorobjekt, som du kan gennemløbe ved hjælp af asynkron iteration, metoden next() eller hændelseslyttere. Streamen forbliver åben og leverer hændelser, efterhånden som de opstår. Et tomt kald til watch() uden en pipeline modtager alle typer af ændringshændelser på samlingen.
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;
}Forbrug hændelser med asynkron iteration
Den mest læsbare måde at forbruge hændelser fra change streams på i moderne Node.js er med asynkron iteration ved hjælp af for await...of. Denne syntaks håndterer automatisk cursorens kald til next() og sætter løkken på pause mellem hændelser. Løkken kører uendeligt, indtil change streamen lukkes, eller der opstår en fejl. Pak altid løkken ind i en try/finally for at sikre, at streamen lukkes, når din proces afsluttes.
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();
}
}Forbrug hændelser med EventEmitter
Alternativt kan du bruge change streamen som en Node.js-EventEmitter. Denne tilgang er velkendt, hvis du bruger streams andre steder i din kode. Registrer en 'change'-hændelseshandler til normale hændelser og en 'error'-handler til forbindelsesproblemer. Denne stil er nyttig, når du vil reagere på hændelser uden at blokere en funktion med en await-løkke.
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();Omfang for change streams: samling, database, klient
Du kan åbne change streams på tre omfang: samlingsniveau (overvåger én samling), databaseniveau (overvåger alle samlinger i en database) og klientniveau (overvåger alle databaser og samlinger i installationen). Jo bredere omfanget er, desto større bliver mængden af hændelser. De fleste applikationer overvåger bestemte samlinger for kun at modtage de hændelser, de har brug for.
// 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);
}Indstillinger for change streams: fullDocument
Som standard indeholder update-hændelser kun felterne, der blev ændret (opdateringsbeskrivelsen), ikke hele dokumentet. Hvis du har brug for hele det opdaterede dokument i hændelsesdataene, skal du sende { fullDocument: 'updateLookup' } til watch(). Det får MongoDB til at foretage et ekstra opslag af dokumentet efter opdateringen og inkludere det i ændringshændelsen. Vær opmærksom på, at dette tilføjer latenstid og er en separat læsning efter hændelsen.
// 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);
}
}Anvendelse: Dashboard i realtid
Et klassisk anvendelsesområde for change streams er at drive et live-dashboard, der viser nye ordrer, efterhånden som de ankommer. Når et nyt ordredokument indsættes, udløses change streamen, og din Node.js-backend kan sende opdateringen til forbundne klienter via WebSocket eller Server-Sent Events. Det eliminerer polling og giver ægte realtidsopdateringer uden at overbelaste databasen med gentagne forespørgsler.
// 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
});
}
}Hændelsesdrevne mikrotjenester med change streams
Change streams kan erstatte en meddelelsesmægler som Kafka eller RabbitMQ i enkle hændelsesdrevne mønstre for mikrotjenester. Når Service A skriver til MongoDB, overvåger Service B samlingen og reagerer på ændringer. Det undgår den driftsmæssige kompleksitet ved en separat meddelelsesbus ved små til mellemstore hændelsesmængder. Ved anvendelser med høj gennemstrømning giver dedikerede meddelelsesmæglere dog bedre garantier og højere gennemstrømning end 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);
}
}
}Luk et change stream korrekt
Change streams bruger en vedvarende forbindelse til MongoDB-serveren. Luk dem altid, når din proces lukker ned, eller når du ikke længere har brug for streamen. Kald changeStream.close(), som returnerer et Promise. I langvarige applikationer skal du lytte efter processignaler (SIGTERM, SIGINT) og lukke streams og klienter, før processen afsluttes, så ressource-lækager på serversiden undgås.
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 kontra polling
Før change streams var applikationer afhængige af polling – de forespurgte regelmæssigt MongoDB for at opdage nye eller opdaterede dokumenter. Polling spilder ressourcer (forespørgsler kører, selv når intet er ændret), medfører latenstid (forsinkelsen ved registrering svarer til polling-intervallet) og skalerer dårligt ved høje hændelsesrater. Change streams eliminerer polling fuldstændigt: Din applikation får øjeblikkeligt besked, når data ændres, og bruger kun minimale ressourcer, når samlingen er inaktiv. Derfor er change streams det foretrukne mønster til alle funktioner, der skal reagere på dataændringer.
Hurtigt tjek
Test din forståelse af begreberne i MongoDB & NoSQL Databases fra denne lektion.
Opsummering af lektionen
I denne lektion har du lært, at change streams leverer et realtidsbaseret hændelsesfeed med alle CRUD-operationer på en samling, database eller installation, at de kræver et replica set og forbruges via asynkron iteration eller EventEmitter, og at indstillingen fullDocument anmoder om hele det opdaterede dokument i update-hændelser. Som det næste ser vi på strukturen af dokumenter med ændringshændelser og på, hvordan du håndterer hver operationstype.
Lær JavaScript med en AI-underviser — gratis
Skriv og kør rigtig kode i din browser, få øjeblikkelig hjælp fra en AI-underviser døgnet rundt, og fortsæt, hvor du slap, på web eller i appen.
- Kurser
- 30
- Lektioner
- 120
Ofte stillede spørgsmål
Er lektionen “Åbning af en ændringsstrøm på en samling” gratis?
Ja — hele teksten til “Åbning af en ændringsstrøm på en samling” kan læses gratis her på nettet. Hvis du vil øve dig interaktivt med en indbygget kodeeditor og en AI-vejleder døgnet rundt og få adgang til resten af MongoDB Academy-kurset, skal du opgradere til CoddyKit PRO. MongoDB Academy-kurset indeholder 4 lektioner i alt.
Hvad lærer jeg i “Åbning af en ændringsstrøm på en samling”?
De lærende vil kalde watch() på en samling og modtage hændelsesstrømmen i en Node.js-applikation ved hjælp af asynkron iteration. Du øver dig i MongoDB Academy med praktisk kode, som du kører direkte i browseren, og en AI-vejleder døgnet rundt besvarer dine spørgsmål, mens du arbejder dig gennem lektionen.
Skal jeg have erfaring for at begynde på MongoDB Academy?
Der kræves ingen tidligere erfaring. MongoDB Academy på CoddyKit er tilrettelagt for både begyndere og øvede, så du kan starte her eller fra begyndelsen og lære i dit eget tempo. Dette er lektion 1 af 4.
Hvor lang tid tager lektionen “Åbning af en ændringsstrøm på en samling”?
De fleste CoddyKit-lektioner tager cirka 5–10 minutter. Hver lektion er kort og interaktiv, så du gør løbende fremskridt og kan fortsætte, hvor du slap – på både web og app.
Kan jeg skrive og køre kode i denne MongoDB Academy-lektion?
Ja. Alle MongoDB Academy-lektioner har en indbygget kodeeditor, så du kan skrive og køre rigtig kode direkte i din browser og få øjeblikkelig feedback fra AI – uden lokal opsætning.
Alle lektioner i dette kursus
- Åbning af en ændringsstrøm på en samling
- Strukturen af et ændringshændelsesdokument
- Filtrering af hændelser med en aggregeringspipeline
- Genoptagelse af ændringsstrømme efter afbrydelser