Apertura de un flujo de cambios en una colección
Llamará a watch() en una colección y consumirá el flujo de eventos en una aplicación de Node.js mediante iteración asíncrona.
Apertura de un flujo de cambios en una colección es una lección gratuita de MongoDB Academy en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de MongoDB Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de MongoDB Academy incluye 4 lecciones en total.
¿Qué son los change streams?
Los change streams proporcionan un flujo de eventos en tiempo real de todas las operaciones de inserción, actualización, reemplazo, eliminación e invalidación realizadas en una colección, base de datos o implementación completa de MongoDB. Se introdujeron en MongoDB 3.6 y se basan en el oplog (registro de operaciones), el diario de replicación del conjunto de réplicas, pero exponen una API de cursor de alto nivel que se puede reanudar, en lugar de exigirle analizar el formato sin procesar del oplog.
Requisitos previos para los change streams
Los change streams requieren un conjunto de réplicas o un clúster fragmentado; no funcionan en instancias independientes de MongoDB porque dependen del oplog. En MongoDB Atlas, todos los clústeres (incluso el nivel gratuito M0) son conjuntos de réplicas, por lo que los change streams funcionan desde el primer momento. Para el desarrollo local, debe iniciar mongod con --replSet rs0 e iniciar el conjunto de réplicas con rs.initiate() en 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()Abrir un change stream con watch()
Llame a collection.watch() para abrir un change stream en una colección específica. El método devuelve un objeto cursor ChangeStream que puede recorrer mediante iteración asíncrona, el método next() o escuchas de eventos. El flujo permanece abierto y entrega los eventos a medida que ocurren. Una llamada a watch() vacía, sin ninguna canalización, recibe todos los tipos de eventos de cambio de esa colección.
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;
}Consumir eventos mediante iteración asíncrona
La forma más legible de consumir eventos de change streams en Node.js moderno es utilizar la iteración asíncrona con for await...of. Esta sintaxis gestiona automáticamente las llamadas next() del cursor y pausa el bucle entre eventos. El bucle se ejecuta indefinidamente hasta que se cierra el change stream o se produce un error. Incluya siempre el bucle en un bloque try/finally para asegurarse de que el flujo se cierre cuando finalice el proceso.
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();
}
}Consumir eventos con EventEmitter
Como alternativa, puede utilizar el change stream como un EventEmitter de Node.js. Este enfoque le resultará familiar si utiliza streams en otras partes del código. Registre un controlador para el evento 'change' para los eventos normales y otro para 'error' para los problemas de conexión. Este estilo resulta útil cuando desea reaccionar a los eventos sin bloquear una función con un bucle await.
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();Ámbito del change stream: colección, base de datos, cliente
Puede abrir change streams en tres ámbitos: nivel de colección (observa una colección), nivel de base de datos (observa todas las colecciones de una base de datos) y nivel de cliente (observa todas las bases de datos y colecciones de la implementación). Cuanto más amplio sea el ámbito, mayor será el volumen de eventos. La mayoría de las aplicaciones observan colecciones específicas para recibir únicamente los eventos que les interesan.
// 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);
}Opciones del change stream: fullDocument
De forma predeterminada, los eventos de actualización solo incluyen los campos que han cambiado (la descripción de la actualización), no el documento completo. Si necesita el documento completo actualizado en la carga útil del evento, pase { fullDocument: 'updateLookup' } a watch(). Esto hace que MongoDB realice una búsqueda adicional del documento después de la actualización y lo incluya en el evento de cambio. Tenga en cuenta que esto añade latencia y constituye una lectura independiente posterior al evento.
// 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);
}
}Caso de uso: panel en tiempo real
Un caso de uso clásico de los change streams es alimentar un panel en vivo que muestre los pedidos nuevos a medida que llegan. Cuando se inserta un documento de pedido nuevo, se activa el change stream y su backend de Node.js puede enviar la actualización a los clientes conectados mediante WebSocket o Server-Sent Events. Esto elimina el sondeo y proporciona actualizaciones realmente en tiempo real sin sobrecargar la base de datos con consultas repetidas.
// 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
});
}
}Microservicios basados en eventos con change streams
Los change streams pueden sustituir a un intermediario de mensajes como Kafka o RabbitMQ en patrones sencillos de microservicios basados en eventos. Cuando el Servicio A escribe en MongoDB, el Servicio B observa la colección y reacciona a los cambios. Esto evita la complejidad operativa de un bus de mensajes independiente cuando el volumen de eventos es bajo o medio. Sin embargo, para casos de uso de alto rendimiento, los intermediarios de mensajes específicos ofrecen mejores garantías y un mayor rendimiento que los 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);
}
}
}Cerrar un change stream correctamente
Los change streams consumen una conexión persistente con el servidor de MongoDB. Ciérrelos siempre cuando se apague el proceso o cuando ya no necesite el flujo. Llame a changeStream.close(), que devuelve una Promise. En aplicaciones de larga duración, escuche las señales del proceso (SIGTERM, SIGINT) y cierre los flujos y los clientes antes de salir para evitar fugas de recursos en el servidor.
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 frente a sondeo
Antes de los change streams, las aplicaciones dependían del sondeo: consultaban periódicamente MongoDB para detectar documentos nuevos o actualizados. El sondeo desperdicia recursos (las consultas se ejecutan incluso cuando no ha cambiado nada), introduce latencia (el retraso de detección equivale al intervalo de sondeo) y se escala mal con tasas de eventos elevadas. Los change streams eliminan por completo el sondeo: su aplicación recibe una notificación inmediata cuando cambian los datos y consume recursos mínimos cuando la colección está inactiva. Por ello, los change streams son el patrón preferido para cualquier funcionalidad que deba reaccionar a cambios en los datos.
Comprobación rápida
Compruebe su comprensión de los conceptos de MongoDB y las bases de datos NoSQL de esta lección.
Resumen de la lección
En esta lección ha aprendido que los change streams proporcionan un flujo de eventos en tiempo real de todas las operaciones CRUD realizadas en una colección, base de datos o implementación; que requieren un conjunto de réplicas y se consumen mediante iteración asíncrona o EventEmitter; y que la opción fullDocument solicita el documento completo actualizado en los eventos de actualización. A continuación, exploraremos la estructura de los documentos de eventos de cambio y cómo gestionar cada tipo de operación.
Aprende JavaScript con un tutor de IA — gratis
Escribe y ejecuta código real en tu navegador, obtén ayuda instantánea de un tutor de IA disponible 24/7 y continúa donde lo dejaste en la web o en la aplicación.
- Cursos
- 30
- Lecciones
- 120
Preguntas frecuentes
¿La lección «Apertura de un flujo de cambios en una colección» es gratis?
Sí — el texto completo de «Apertura de un flujo de cambios en una colección» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de MongoDB Academy, actualiza a CoddyKit PRO. El curso de MongoDB Academy incluye 4 lecciones en total.
¿Qué aprenderé en «Apertura de un flujo de cambios en una colección»?
Llamará a watch() en una colección y consumirá el flujo de eventos en una aplicación de Node.js mediante iteración asíncrona. Practicas MongoDB Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar MongoDB Academy?
No se requiere experiencia previa. MongoDB Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.
¿Cuánto tiempo toma la lección «Apertura de un flujo de cambios en una colección»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de MongoDB Academy?
Sí. Cada lección de MongoDB Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Apertura de un flujo de cambios en una colección
- Estructura del documento de un evento de cambio
- Filtrado de eventos con un pipeline de agregación
- Reanudación de flujos de cambios tras una interrupción