Para agentes de IA: um índice de documentação está disponível em https://www.mongodb.com/pt-br/docs/llms.txt — as versões de markdown de todas as páginas estão disponíveis anexando .md a qualquer caminho de URL.
Menu Docs
Página inicial do Docs
/ /
Driver de fluxos reativos do Java
Página inicial do Docs
/
Bibliotecas do cliente
/
Java
/
Driver de fluxos reativos do Java
Página inicial do Docs
/
Bibliotecas do cliente
/
Java
/
Driver de fluxos reativos do Java

Fluxos de alterações

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.

Dados de amostra

Os exemplos neste guia usam a collection sample_restaurants.restaurants dos conjuntos de dados de amostra do Atlas. Para saber como criar um cluster gratuito do MongoDB Atlas e carregar os conjuntos de dados de amostra, consulte o guia de Introdução .

Importante

Biblioteca do Reator do Projeto

Este guia usa a biblioteca Project Reactor para consumir instâncias do Publisher retornadas pelos métodos de driver Java Reactive Streams. Para saber mais sobre a biblioteca do Projeto Reactor e como usá-la, consulte Introdução na documentação do Reactor. Para saber mais sobre como usamos os métodos da biblioteca do Project Reactor neste guia, consulte o guia Gravar dados no MongoDB.

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:

Opção
Descrição

fullDocument()

Especifica se o documento completo deve ser mostrado após a alteração, em vez de mostrar apenas as alterações feitas no documento. Para saber mais sobre essa opção, consulte Incluir pré-imagens e pós-imagens.

fullDocumentBeforeChange()

Especifica se o documento completo deve ser mostrado como estava antes da alteração, em vez de mostrar apenas as alterações feitas no documento. Para saber mais sobre essa opção, consulte Incluir pré-imagens e pós-imagens.

resumeAfter()

Direciona watch() para retomar as alterações de retorno após a operação especificada no token de retomada.
Cada documento de evento de fluxo de alteração inclui um token de currículo como o _id campo. Passe todo o _id campo do documento do evento de alteração que representa a operação que você deseja retomar depois.
resumeAfter() é mutuamente exclusivo com startAfter() e.startAtOperationTime()

startAfter()

Direciona watch() para iniciar um novo fluxo de alterações após a operação especificada no token de retomada. Permite que as notificações sejam retomadas após um evento de invalidação.
Cada documento de evento de fluxo de alteração inclui um token de currículo como o _id campo. Passe todo o _id campo do documento do evento de alteração que representa a operação que você deseja retomar depois.
startAfter() é mutuamente exclusivo com resumeAfter() startAtOperationTime()e.

startAtOperationTime()

Direciona watch() para retornar apenas eventos que ocorrem após o registro de data/hora especificado.
startAtOperationTime() é mutuamente exclusivo com resumeAfter() startAfter()e.

maxAwaitTime()

Especifica a quantidade máxima de tempo, em milissegundos, o servidor aguarda novas alterações de dados para relatar ao cursor do fluxo de alterações antes de retornar um lote vazio. Padrão para 1000 milissegundos.

showExpandedEvents()

A partir do MongoDB Server v6.0, Os fluxos de alterações oferecem suporte a notificações de alteração para eventos de Linguagem de Definição de Dados (DDL), como os eventos createIndexes e dropIndexes. Para incluir eventos expandidos em um change stream, chame esse método e passe o valor true.

batchSize()

Especifica o número máximo de eventos de alteração a serem retornados em cada lote da resposta do cluster MongoDB . Por padrão, o driver define esse valor como Long.MAX_VALUE.

Um batchSize de 0 significa que o cursor será estabelecido, mas nenhum documento será retornado no primeiro lote.

collation()

Especifica o agrupamento a ser usado para o cursor do change stream.

comment()

Anexa um comentário à operação.

Incluir pré-imagens e pós-imagens

Importante

Você pode habilitar pré-imagens e pós-imagens em collections somente se seu sistema usar MongoDB v6.0 ou posterior.

Por padrão, quando você executa uma operação em uma collection, o evento de alteração correspondente inclui somente o delta dos campos modificados por essa operação. Para ver o documento completo antes ou depois de uma alteração, encadeie o método fullDocumentBeforeChange() ou fullDocument() ao método watch() .

A pré-imagem é a versão completa de um documento antes de uma alteração. Para incluir a pré-imagem no evento de fluxo de alteração, passe um dos seguintes valores para o método fullDocumentBeforeChange() :

  • FullDocumentBeforeChange.WHEN_AVAILABLE: o evento de alteração inclui uma pré-imagem do documento modificado para eventos de alteração somente se a pré-imagem estiver disponível.

  • FullDocumentBeforeChange.REQUIRED: o evento de alteração inclui uma pré-imagem do documento modificado para eventos de alteração. Se a pré-imagem não estiver disponível, o driver gerará um erro.

A pós-imagem é a versão completa de um documento após uma alteração. Para incluir a pós-imagem no evento de fluxo de alteração, passe um dos seguintes valores para o método fullDocument() :

  • FullDocument.UPDATE_LOOKUP: o evento de alteração inclui uma cópia de todo o documento alterado de algum tempo após a alteração.

  • FullDocument.WHEN_AVAILABLE: O evento de alteração inclui uma pós-imagem do documento modificado para eventos de alteração somente se a pós-imagem estiver disponível.

  • FullDocument.REQUIRED: o evento de alteração inclui uma pós-imagem do documento modificado para eventos de alteração. Se a pós-imagem não estiver disponível, o driver gerará um erro.

O exemplo a seguir abre um fluxo de alteração em uma coleção e inclui a pós-imagem de documentos atualizados encadeando o método fullDocument() ao método watch() :

// 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();

Para saber mais sobre pré e pós-imagens, consulte Change Streams com pré e pós-imagens de documentos no manual do MongoDB Server .

Informações adicionais

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

Documentação da API

Para saber mais sobre qualquer um dos métodos ou tipos discutidos neste guia, consulte a seguinte documentação da API:

Voltar

Monitoramento

Próximo

Autenticação

Nesta página