Fluxos de change streams

Visão geral

Neste guia, você pode aprender como usar um change stream para monitorar alterações em tempo real em seu reconhecimento de data center. Um change stream é um recurso do MongoDB Server que permite que sua aplicação assine alterações de dados em uma única collection, reconhecimento de data center ou sistema.

Você pode especificar um conjunto de operadores de agregação para filtrar e transformar os dados que seu aplicativo recebe. Ao se conectar a um sistema MongoDB v6.0 ou posterior, você também pode configurar os eventos para incluir os dados do documento antes e depois da alteração.

Saiba como abrir e configurar seu change stream nas seguintes seções:

• Abrir um fluxo de alterações

• Aplique operadores de agregação ao seu change stream

• Incluir pré-imagens e pós-imagens

Abrir um fluxo de alterações

Você pode abrir um fluxo de alterações para assinar tipos específicos de alterações de dados e produzir eventos de alteração em seu aplicativo.

Para abrir um change stream, chame o método watch() em uma instância de um MongoCollection, MongoDatabase ou MongoClient.

O objeto no qual você chama o método watch() determina o escopo de eventos que o change stream escuta.

Se você chamar watch() em um MongoCollection, o change stream monitorará uma collection.

Se você chamar watch() em um MongoDatabase, o change stream monitorará todas as collection nesse reconhecimento de data center.

Se você chamar watch() em um MongoClient , o change stream monitorará todas as alterações na implantação do MongoDB conectada.

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch();
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=insert,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

Aplique operadores de agregação ao seu change stream

Você pode passar um pipeline de agregação como um parâmetro para o método watch() para especificar quais eventos de alteração o fluxo de alteração recebe.

Para saber quais operadores de aggregation sua versão do MongoDB Server suporta, consulte Modificar a saída do change stream.

MongoCollection<Document> collection = database.getCollection("myColl");
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.in("operationType", Arrays.asList("insert", "update"))));
ChangeStreamIterable<Document> changeStream = collection.watch(pipeline);
changeStream.forEach(event ->
        System.out.println("Received a change to the collection: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

Dividir eventos de fluxo de grandes mudanças

A partir do MongoDB 7.0, você pode usar o estágio de agregação $changeStreamSplitLargeEvent para fazer a divisão de evento que excedem 16 MB em fragmentos menores.

Utilize o $changeStreamSplitLargeEvent somente quando estritamente necessário. For example, use $changeStreamSplitLargeEvent if your application requires full document pre- or post-images, and generates events that exceed 16 MB.

O estágio $changeStreamSplitLargeEvent retorna os fragmentos sequencialmente. Você pode acessar os fragmentos usando um cursor de change stream. Cada fragmento inclui um objeto SplitEvent contendo os seguintes campos:

O exemplo a seguir modifica seu change stream usando o estágio de aggregation $changeStreamSplitLargeEvent para fazer a divisão de evento grandes:

ChangeStreamIterable<Document> changeStream = collection.watch(
      Arrays.asList(Document.parse("{ $changeStreamSplitLargeEvent: {} }")));

Você pode chamar o método getSplitEvent() no cursor do change stream para acessar o SplitEvent , conforme mostrado no exemplo a seguir:

MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = changeStream.cursor();
SplitEvent event = cursor.tryNext().getSplitEvent();

Para obter mais informações sobre o estágio de agregação $changeStreamSplitLargeEvent , consulte a documentação do servidor $changeStreamSplitLargeEvent .

Incluir pré-imagens e pós-imagens

Você pode configurar o evento de alteração para conter ou omitir os seguintes dados:

• A pré-imagem, um documento que representa a versão do documento antes da operação, se existir

• A pós-imagem, um documento que representa a versão do documento após a operação, se existir

Para receber evento de change stream que incluem uma pré-imagem ou pós-imagem, você deve executar a seguinte ação:

• Habilite pré-imagens e pós-imagens para a coleção em seu sistema do MongoDB.

  Dica

  Para saber como ativar as pré e pós-imagens em sua implantação, consulte Fluxo de alterações com pré e pós-imagens de documentos no manual do servidor.

  Para saber como instruir o driver a criar uma collection com pré-imagens e pós-imagens habilitadas, consulte a seção Criar uma collection com pré-imagem e pós-imagens habilitadas .

• Configure seu change stream para recuperar as pré-imagens e as pós-imagens ou as duas imagens.

  Dica

  Para configurar seu change stream para registrar a pré-imagem em eventos de alteração, consulte o Exemplo de configuração de pré-imagem.

  Para configurar seu change stream para registrar a pós-imagem em eventos de alteração, consulte o Exemplo de configuração pós-imagem.

CreateCollectionOptions collectionOptions = new CreateCollectionOptions();
collectionOptions.changeStreamPreAndPostImagesOptions(new ChangeStreamPreAndPostImagesOptions(true));
database.createCollection("myColl", collectionOptions);

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocumentBeforeChange(FullDocumentBeforeChange.REQUIRED);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=null,
   fullDocumentBeforeChange=Document{{_id=..., amount=150, ...}},
   ...
}

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocument(FullDocument.WHEN_AVAILABLE);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=Document{{_id=..., color=purple, ...}},
   updatedFields={"color": purple},
   ...
}