Monitorar alterações de dados

Visão geral

Neste guia, você pode aprender como usar um fluxo de alterações para monitorar alterações em tempo real em seus dados. Um change stream é uma funcionalidade do MongoDB Server que permite que seu aplicação se inscreva em alterações de dados em uma collection, banco de dados de dados ou sistema.

Abrir um fluxo de alterações

Para abrir um fluxo de alteração, chame o método watch() . A instância na qual você chama o método determina o escopo de eventos que o change stream escuta. Você pode chamar o método watch() nas instâncias das seguintes classes:

• MongoClient: Para monitorar todas as alterações no sistema MongoDB

• MongoDatabase: Para monitorar alterações em todas as coleções no banco de dados

• MongoCollection: Para monitorar alterações na coleção

O exemplo a seguir abre um fluxo de alteração na coleção restaurants e gera as alterações conforme elas ocorrem:

// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch();
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Para começar a observar as alterações, execute o aplicação. Em seguida, em um aplicação ou shell separado, execute uma operação de gravação na coleção restaurants . Atualizar um documento que tenha um valor de campo "name" de "Blarney Castle" resulta na seguinte saída do fluxo de alteração:

Received change: ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null, fullDocument=null,
fullDocumentBeforeChange=null, documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...},
updateDescription=UpdateDescription{removedFields=[], updatedFields={"cuisine": "Traditional Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{value=...}}

Modificar a saída change stream

Você pode passar um pipeline de agregação como um parâmetro para o método watch() para modificar a saída do fluxo de alteração. Esse parâmetro permite que você observe somente eventos de alteração especificados.

Você pode especificar os seguintes estágios de agregação no parâmetro pipeline :

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

O exemplo a seguir passa um pipeline de agregação para um fluxo de alterações para registrar somente as operações de atualização:

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Para saber mais sobre como modificar a saída do change stream, consulte a seção Modificar a saída do change stream no manual do MongoDB Server .

Modificar Comportamento do watch()

Você pode encadear métodos para o método watch() que representam opções que você pode usar para configurar a operação de change stream . Se você não especificar nenhuma opção, o driver não personalizará a operação.

A tabela a seguir descreve os métodos que você pode conectar a watch() para personalizar seu comportamento:

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received including the full
// document after the update
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

Informações adicionais

Para saber mais sobre fluxos de alterações, consulte Change Streams de alterações no manual do MongoDB Server .