Abrindo um fluxo de alterações em uma coleção
Você chamará watch() em uma coleção e consumirá o fluxo de eventos em uma aplicação Node.js usando iteração assíncrona.
Abrindo um fluxo de alterações em uma coleção é uma aula grátis de MongoDB Academy no CoddyKit. Esta é a aula 1 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de MongoDB Academy, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de MongoDB Academy inclui 4 aulas no total.
O que são fluxos de alterações?
Os fluxos de alterações fornecem um fluxo de eventos em tempo real com todas as operações de inserção, atualização, substituição, exclusão e invalidação em uma coleção, banco de dados ou implantação inteira do MongoDB. Introduzidos no MongoDB 3.6, eles são baseados no oplog (registro de operações) — o diário de replicação do conjunto de réplicas —, mas expõem uma API de cursor de alto nível que pode ser retomada, em vez de exigir que você analise o formato bruto do oplog.
Pré-requisitos dos fluxos de alterações
Os fluxos de alterações exigem um conjunto de réplicas ou cluster fragmentado — eles não funcionam em instâncias autônomas do MongoDB porque dependem do oplog. No MongoDB Atlas, todos os clusters (até mesmo o nível gratuito M0) são conjuntos de réplicas, portanto os fluxos de alterações funcionam imediatamente. Para o desenvolvimento local, você precisa iniciar o mongod com --replSet rs0 e iniciar o conjunto de réplicas com rs.initiate() no 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()Abrindo um fluxo de alterações com watch()
Chame collection.watch() para abrir um fluxo de alterações em uma coleção específica. O método retorna um objeto cursor ChangeStream que você pode percorrer usando iteração assíncrona, o método next() ou ouvintes de eventos. O fluxo permanece aberto e entrega eventos à medida que eles ocorrem. Uma chamada watch() vazia, sem pipeline, recebe todos os tipos de eventos de alteração dessa coleção.
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;
}Consumindo eventos com iteração assíncrona
A maneira mais legível de consumir eventos de fluxos de alterações no Node.js moderno é usar a iteração assíncrona com for await...of. Essa sintaxe trata automaticamente das chamadas next() do cursor e pausa o laço entre os eventos. O laço é executado indefinidamente até que o fluxo de alterações seja fechado ou ocorra um erro. Sempre envolva o laço em um try/finally para garantir que o fluxo seja fechado quando seu processo for encerrado.
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();
}
}Consumindo eventos com EventEmitter
Como alternativa, você pode usar o fluxo de alterações como um EventEmitter do Node.js. Essa abordagem é familiar se você usa fluxos em outras partes do seu código. Registre um manipulador para o evento 'change' para eventos normais e um manipulador para 'error' para problemas de conexão. Esse estilo é útil quando você quer reagir a eventos sem bloquear uma função com um laço 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();Escopo do fluxo de alterações: coleção, banco de dados e cliente
Você pode abrir fluxos de alterações em três escopos: nível da coleção (observa uma coleção), nível do banco de dados (observa todas as coleções de um banco de dados) e nível do cliente (observa todos os bancos de dados e coleções da implantação). Quanto mais amplo o escopo, maior o volume de eventos. A maioria das aplicações observa coleções específicas para receber apenas os eventos relevantes.
// 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);
}Opções do fluxo de alterações: fullDocument
Por padrão, os eventos de atualização incluem apenas os campos que foram alterados (a descrição da atualização), não o documento inteiro. Se precisar do documento completo atualizado na carga do evento, passe { fullDocument: 'updateLookup' } para watch(). Isso faz com que o MongoDB faça uma busca adicional do documento após a atualização e o inclua no evento de alteração. Observe que isso adiciona latência e constitui uma leitura separada após o 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: painel em tempo real
Um caso de uso clássico dos fluxos de alterações é alimentar um painel ao vivo que mostre novos pedidos à medida que chegam. Quando um novo documento de pedido é inserido, o fluxo de alterações é acionado, e seu back-end Node.js pode enviar a atualização aos clientes conectados por meio de WebSocket ou eventos enviados pelo servidor. Isso elimina a sondagem e fornece atualizações realmente em tempo real sem sobrecarregar o banco de dados com 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
});
}
}Microsserviços orientados a eventos com fluxos de alterações
Os fluxos de alterações podem substituir um intermediário de mensagens como Kafka ou RabbitMQ em padrões simples de microsserviços orientados a eventos. Quando o Serviço A grava no MongoDB, o Serviço B observa a coleção e reage às alterações. Isso evita a complexidade operacional de um barramento de mensagens separado para volumes de eventos baixos ou médios. No entanto, para casos de uso de alta vazão, os intermediários de mensagens dedicados oferecem garantias melhores e uma vazão maior que os fluxos de alterações.
// 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);
}
}
}Fechando um fluxo de alterações de forma adequada
Os fluxos de alterações consomem uma conexão persistente com o servidor MongoDB. Sempre os feche quando seu processo for encerrado ou quando você não precisar mais do fluxo. Chame changeStream.close(), que retorna uma Promise. Em aplicações de longa duração, escute os sinais do processo (SIGTERM, SIGINT) e feche os fluxos e clientes antes de sair para evitar vazamentos de recursos no lado do 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);
});Fluxos de alterações versus sondagem
Antes dos fluxos de alterações, as aplicações dependiam da sondagem — consultas periódicas ao MongoDB para detectar documentos novos ou atualizados. A sondagem desperdiça recursos (as consultas são executadas mesmo quando nada mudou), introduz latência (o atraso da detecção é igual ao intervalo da sondagem) e não escala bem com altas taxas de eventos. Os fluxos de alterações eliminam completamente a sondagem: sua aplicação é notificada imediatamente quando os dados mudam, consumindo recursos mínimos quando a coleção está inativa. Isso faz dos fluxos de alterações o padrão preferido para qualquer recurso que precise reagir a mudanças nos dados.
Verificação rápida
Teste sua compreensão dos conceitos de MongoDB e bancos de dados NoSQL desta lição.
Recapitulação da lição
Nesta lição, você aprendeu que: os fluxos de alterações fornecem um fluxo de eventos em tempo real com todas as operações CRUD em uma coleção, banco de dados ou implantação; eles exigem um conjunto de réplicas e são consumidos por iteração assíncrona ou EventEmitter; e a opção fullDocument solicita o documento completo atualizado nos eventos de atualização. A seguir, exploraremos a estrutura dos documentos de eventos de alteração e como tratar cada tipo de operação.
Aprenda JavaScript com um tutor de IA — grátis
Escreva e execute código real no seu navegador, obtenha ajuda instantânea de um tutor de IA 24/7 e continue de onde parou na web ou no app.
- Cursos
- 30
- Aulas
- 120
Perguntas Frequentes
A aula “Abrindo um fluxo de alterações em uma coleção” é grátis?
Sim — o texto completo de “Abrindo um fluxo de alterações em uma coleção” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de MongoDB Academy, atualize para CoddyKit PRO. O curso de MongoDB Academy inclui 4 aulas no total.
O que vou aprender em “Abrindo um fluxo de alterações em uma coleção”?
Você chamará watch() em uma coleção e consumirá o fluxo de eventos em uma aplicação Node.js usando iteração assíncrona. Você pratica MongoDB Academy com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.
Preciso ter experiência prévia para começar MongoDB Academy?
Nenhuma experiência prévia é necessária. MongoDB Academy no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 1 de 4.
Quanto tempo leva a aula “Abrindo um fluxo de alterações em uma coleção”?
A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.
Posso escrever e executar código nesta aula de MongoDB Academy?
Sim. Cada aula de MongoDB Academy inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.
Todas as aulas deste curso
- Abrindo um fluxo de alterações em uma coleção
- Estrutura do documento de evento de alteração
- Filtrando eventos com um pipeline de agregação
- Retomando fluxos de alterações após uma interrupção