Menu Docs
Página inicial do Docs
/
Idiomas
/
Java
/
Driver de fluxos reativos do Java
/
Tutorials

Fluxos de alterações

Nesta página

  • Pré-requisitos
  • Conecte-se a um MongoDB deployment
  • Fique atento às mudanças em uma coleção
  • Fique atento às mudanças em um banco de dados
  • Fique atento às alterações em todos os bancos de dados
  • Filtrando conteúdo

O MongoDB 3.6 apresenta o operador de pipeline de agregação $changeStream .

Os change streams fornecem uma maneira de observar as alterações nos documentos de uma coleção. Para melhorar a usabilidade desta nova etapa, o tipo MongoCollection inclui um novo método watch() . A instância ChangeStreamPublisher configura o change stream e tenta retomar automaticamente se encontrar um erro potencialmente recuperável.

Pré-requisitos

Você deve configurar os seguintes componentes para executar os exemplos de código neste guia:

  • Uma test.restaurants coleção preenchida com documentos do restaurants.json arquivo nos ativos Github de documentação do .

  • As seguintes declarações de importação:

import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import org.bson.Document;

Importante

Este guia usa as implementações do Subscriber , que são descritas noPrimário de início rápidodo .

Conecte-se a um MongoDB deployment

Primeiro, conecte a um MongoDB deployment e, em seguida, declare e defina as instâncias MongoDatabase e MongoCollection .

O código a seguir se conecta a uma MongoDB deployment standalone em execução em localhost na porta 27017. Em seguida, define a variável database para fazer referência ao banco de dados test e a variável collection para fazer referência à coleção restaurants :

MongoClient mongoClient = MongoClients.create();
MongoDatabase database = mongoClient.getDatabase("test");
MongoCollection<Document> collection = database.getCollection("restaurants");

Para saber mais sobre como se conectar a sistemas do MongoDB, consulte o tutorial Conectar ao MongoDB .

Fique atento às mudanças em uma coleção

Para criar um change stream, use um dos métodos MongoCollection.watch() .

No exemplo a seguir , o fluxo de alterações imprime todas as alterações observadas:

collection.watch().subscribe(new PrintDocumentSubscriber());

Fique atento às mudanças em um banco de dados

Os aplicativos podem abrir um único change stream para observar todas as collections que não são do sistema de um banco de dados de dados. Para criar esse change stream, use um dos métodos MongoDatabase.watch() .

No exemplo a seguir, o fluxo de alterações imprime todas as alterações que ele observa no banco de dados fornecido:

database.watch().subscribe(new PrintDocumentSubscriber());

Fique atento às alterações em todos os bancos de dados

Os aplicativos podem abrir um único change stream para observar todas as collections que não são do sistema de todos os bancos de dados em um MongoDB deployment. Para criar esse change stream, use um dos métodos MongoClient.watch() .

No exemplo a seguir, o change stream imprime todas as alterações que observa no sistema ao qual o MongoClient está conectado:

client.watch().subscribe(new PrintDocumentSubscriber());

Filtrando conteúdo

Você pode passar uma lista de estágios de agregação para o método watch() para modificar os dados retornados pelo operador $changeStream .

Observação

Nem todos os operadores de agregação são suportados. Consulte Change Streams no manual do servidor MongoDB para saber mais.

No exemplo a seguir , o change stream imprime todas as alterações que observa correspondentes às operações insert, update, replace e delete .

Primeiro, o pipeline inclui um estágio $match para filtrar documentos em que o operationType é um insert, update, replace ou delete. Em seguida, ele define fullDocument como FullDocument.UPDATE_LOOKUP, para que o documento após a atualização seja incluído nos resultados:

collection.watch(
    asList(
        Aggregates.match(
            Filters.in("operationType", asList("insert", "update", "replace", "delete"))
        )
    )
).fullDocument(FullDocument.UPDATE_LOOKUP).subscribe(new PrintDocumentSubscriber());

Voltar

Framework de aggregation