Java

Atualizações

O repositório de início rápido do MongoDB Java está disponível no GitHub.

28 de fevereiro de 2024

Atualizar para o Java 21
Atualize o driver Java para 5.0.0
Atualize logback-classic para 1.2.13

14 de novembro de 2023

Atualizar para o Java 17
Atualize o driver Java para 4.11.1
Atualize o mongodb-crypt para 1.8.0

25 de março de 2021

Atualize o driver Java para 4.2.2.
Exemplo de criptografia no nível do campo do lado do cliente adicionado.

21 de outubro de 2020

Atualize o driver Java para 4.1.1.
O registro do driver Java agora está ativado por meio da popular APISLF4J, então adicionei logback no pom.xml e um arquivo de configuração logback.xml.

Introdução

Os change stream foram introduzidos no MongoDB 3.6. Eles permitem que os aplicativos acessem alterações de dados em tempo real sem a complexidade e o risco de afetar o oplog.

Os aplicativos podem usar fluxos de alteração para assinar todas as alterações de dados em uma única coleção, um banco de dados ou uma implantação inteira e reagir imediatamente a elas. Como os fluxos de alterações usam a estrutura de agregação, um aplicativo também pode filtrar alterações específicas ou transformar as notificações à vontade.

Nesta publicação do blog, conforme comprometido na primeira publicação do blog desta série, mostrarei como aproveitar os change stream do MongoDB usando Java.

Configuração

Usarei o mesmo repositório de sempre nesta série. Se você ainda não tiver uma cópia dele, poderá cloná-lo ou apenas atualizá-lo, caso já o tenha:

Trecho de código

Se você ainda não configurou seu cluster gratuito no MongoDB Atlas, agora é um ótimo momento para fazê-lo. Você tem todas as instruções nesta postagem do blog.

Fluxos de alterações

Nesta publicação do blog, trabalharei no arquivo chamado ChangeStreams.java, mas é super fácil trabalhar com os Change Streams .

Mostrarei 5 exemplos diferentes para mostrar alguns recursos do change stream. Por uma questão de simplicidade, mostrarei apenas os pedaços de código relacionados diretamente ao change stream. Você pode encontrar toda a amostra de código na parte inferior desta postagem do blog ou no Github.

Para cada exemplo, você precisará iniciar 2 programas Java na ordem correta se quiser reproduzir meus exemplos.

O primeiro programa é sempre aquele que contém o código Change Streams.
O segundo será um dos programas Java que já usamos nesta série de postagens de blog Java. Você pode encontrá-los no repositório do Github. Elas gerarão operações MongoDB que observaremos na saída do change stream.

Um simples Change Streams sem filtros

Vamos começar com o change stream mais simples que podemos criar:

Trecho de código

Como você pode ver, tudo o que precisamos é de myCollection.watch()! É isso aí.

Isso retorna um ChangeStreamIterable que, conforme indicado por seu nome, pode ser iterado para retornar nossos eventos de alteração. Aqui, estou iterando sobre meu Change Stream para imprimir meus documentos de eventos de alteração na saída padrão do Java.

Também posso simplificar este código assim:

Trecho de código

Reutilizarei essa interface funcional nos meus exemplos a seguir para facilitar a leitura.

Para executar este exemplo:

Remova o comentário apenas do exemplo 1 do arquivoChangeStreams.java e inicie-o em seu IDE ou em um console dedicado usando Maven na raiz do seu projeto.

Trecho de código

Inicie MappingPOJO.java em outro console ou no seu IDE.

Trecho de código

No MappingPOJO, estamos realizando 4 operações do MongoDB:

Estou criando um novo documento Gradecom o métodoinsertOne(),
Estou procurando este documentoGrade usando o métodofind(),
Estou substituindo inteiramente este Grade usando o métodofindOneAndReplace(),
e, finalmente, estou excluindo esse Grade usando o métododeleteOne().

Isso é confirmado na saída padrão de MappingJava:

Trecho de código

Vamos verificar o que temos na saída padrão de ChangeStreams.java (prettified):

Trecho de código

ChangeStreamDocument{
   operationType=OperationType{ value='insert' },
   resumeToken={ "_data":"825E2F3E40000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
   namespace=sample_training.grades,
   destinationNamespace=null,
   fullDocument=Grade{
      id=5e2f3e400c47cf19d5936162,
      student_id=10003.0,
      class_id=10.0,
      scores=[ Score { type='homework', score=50.0 } ]
   },
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
   clusterTime=Timestamp{
      value=6786711608069455873,
      seconds=1580154432,
      inc=1
   },
   updateDescription=null,
   txnNumber=null,
   lsid=null
}
ChangeStreamDocument{ operationType=OperationType{ value= 'replace' },
   resumeToken={ "_data":"825E2F3E40000000032B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
   namespace=sample_training.grades,
   destinationNamespace=null,
   fullDocument=Grade{
      id=5e2f3e400c47cf19d5936162,
      student_id=10003.0,
      class_id=10.0,
      scores=[ Score{ type='homework', score=50.0 }, Score{ type='exam', score=42.0 } ]
   },
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
   clusterTime=Timestamp{
      value=6786711608069455875,
      seconds=1580154432,
      inc=3
   },
   updateDescription=null,
   txnNumber=null,
   lsid=null
}
ChangeStreamDocument{
   operationType=OperationType{ value='delete' },
   resumeToken={ "_data":"825E2F3E40000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
   namespace=sample_training.grades,
   destinationNamespace=null,
   fullDocument=null,
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
   clusterTime=Timestamp{
      value=6786711608069455876,
      seconds=1580154432,
      inc=4
   },
   updateDescription=null,
   txnNumber=null,
   lsid=null
}

Como você pode ver, somente 3 operações aparecem no Change Stream:

insert,
substituir,
excluir.

Isso era esperado porque a operaçãofind() é apenas um documento de leitura do MongoDB. Não está mudando nada, portanto, não gerando um evento no Change Stream.

Agora que terminamos o exemplo básico, vamos explorar algumas funcionalidades dos Change Streams.

Encerre o programa Change Stream que iniciamos anteriormente e vamos continuar.

Um simples Change Stream filtrando pelo tipo de operação

Agora vamos fazer a mesma coisa, mas vamos imaginar que estamos interessados apenas em operações de inserção e exclusão.

Trecho de código

Como você pode ver aqui, estou usando o recurso de pipeline de agregação do change stream para filtrar os eventos de mudança que deseja processar.

Remova o comentário do exemplo 2 em ChangeStreams.java e execute o programa seguido por MappingPOJO.java, assim como fizemos anteriormente.

Aqui estão os eventos de mudança que estou recebendo.

Trecho de código

ChangeStreamDocument {operationType=OperationType {value= 'insert'},
  resumeToken= {"_data": "825E2F4983000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F4983CC1D2842BFF555640004"},
  namespace=sample_training.grades,
  destinationNamespace=null,
  fullDocument=Grade
  {
    id=5e2f4983cc1d2842bff55564,
    student_id=10003.0,
    class_id=10.0,
    scores= [ Score {type= 'homework', score=50.0}]
  },
  documentKey= {"_id": {"$oid": "5e2f4983cc1d2842bff55564" }},
  clusterTime=Timestamp {value=6786723990460170241, seconds=1580157315, inc=1 },
  updateDescription=null,
  txnNumber=null,
  lsid=null
}

ChangeStreamDocument { operationType=OperationType {value= 'delete'},
  resumeToken= {"_data": "825E2F4983000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F4983CC1D2842BFF555640004"},
  namespace=sample_training.grades,
  destinationNamespace=null,
  fullDocument=null,
  documentKey= {"_id": {"$oid": "5e2f4983cc1d2842bff55564"}},
  clusterTime=Timestamp {value=6786723990460170244, seconds=1580157315, inc=4},
  updateDescription=null,
  txnNumber=null,
  lsid=null
  }
]

Desta vez, estou recebendo apenas os eventos 2 , insert e delete. O eventoreplace foi filtrado em comparação com o primeiro exemplo.

Alterar o comportamento padrão do Stream com operações de atualização

Igual ao anterior, desta vez estou filtrando meu change stream para manter apenas as operações de atualização.

Trecho de código

Desta vez, siga estas etapas.

remova o comentário do exemplo 3 em ChangeStreams.java,
se você nunca executou Create.java, execute-o. Usaremos esses novos documentos na próxima etapa.
iniciar Update.java em outro console.

Em seu console do change stream, você deverá ver 13 eventos de atualização. Aqui está o primeiro:

Trecho de código

Como você pode ver, estamos recuperando nossa operação de atualização no campoupdateDescription, mas estamos obtendo apenas a diferença com a versão anterior desse documento.

O campofullDocument é null porque, por padrão, o MongoDB apenas envia a diferença para evitar sobrecarregar o change stream com informações potencialmente inúteis.

Vejamos como podemos alterar esse comportamento no próximo exemplo.

Fluxo de alterações com "Pesquisa de atualização"

Para esta parte, descomente o exemplo 4 de ChangeStreams.java e execute os programas conforme acima.

Trecho de código

Adicionei a opção UPDATE_LOOKUP desta vez, para que também possamos recuperar o documento inteiro durante uma operação de atualização.

Vamos ver novamente a primeira atualização no meu fluxo de alterações:

Trecho de código

Observação: o programaUpdate.java atualiza um campo criado "comentários" que não existe em meu POJO Grade, que representa o esquema original desta collection. Assim, o campo não aparece na saída porque não está mapeado.

Se eu quiser ver esse campocomments, posso usar um MongoCollection não mapeado automaticamente para meu POJOGrade.java .

Trecho de código

Então é isso que obtenho no meu change stream:

Trecho de código

Reduzi o campocomments para mantê-lo legível, mas ele contém 14 vezes o mesmo comentário no meu caso.

O documento completo que estamos recuperando aqui durante nossa operação de atualização é o documento após a ocorrência da atualização. Leia mais sobre isso em nossa documentação.

Change Streams são retomáveis

Neste exemplo final 5, simulei um erro e estou reiniciando meu Change Stream a partir de um resumeToken que obtive de uma operação anterior em meu Change Stream.

É importante observar que um fluxo de alterações será retomado automaticamente diante de um incidente "" . Em geral, o único motivo pelo qual um aplicativo precisa reiniciar o fluxo de alterações manualmente a partir de um token de retomada é se houver um incidente no próprio aplicativo e não no fluxo de alterações (por exemplo, um operador decidiu que o aplicativo precisa ser reiniciado).

Trecho de código

private static void exampleWithResumeToken(MongoCollection<Grade> grades) {
    List<Bson> pipeline = List.of(match(eq("operationType", "update")));
    ChangeStreamIterable<Grade> changeStream = grades.watch(pipeline);
    MongoChangeStreamCursor<ChangeStreamDocument<Grade>> cursor = changeStream.cursor();
    System.out.println("==> Going through the stream a first time & record a resumeToken");
    int indexOfOperationToRestartFrom = 5;
    int indexOfIncident = 8;
    int counter = 0;
    BsonDocument resumeToken = null;
    while (cursor.hasNext() && counter != indexOfIncident) {
        ChangeStreamDocument<Grade> event = cursor.next();
        if (indexOfOperationToRestartFrom == counter) {
            resumeToken = event.getResumeToken();
        }
        System.out.println(event);
        counter++;
    }
    System.out.println("==> Let's imagine something wrong happened and I need to restart my Change Stream.");
    System.out.println("==> Starting from resumeToken=" + resumeToken);
    assert resumeToken != null;
    grades.watch(pipeline).resumeAfter(resumeToken).forEach(printEvent());
}

Para este exemplo final, o mesmo que anterior. Descomente a parte 5 (que é apenas chamar o método acima) e comece ChangeStreams.java depois Update.java.

Esta é a saída que você deve obter:

Trecho de código

==> Going through the stream a first time & record a resumeToken
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCCE74AA51A0486763FE0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcce74aa51a0486763fe"}}, clusterTime=Timestamp{value=6786856975532556289, seconds=1580188278, inc=1}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000022B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBA0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbba"}}, clusterTime=Timestamp{value=6786856975532556290, seconds=1580188278, inc=2}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.15": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000032B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBB0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbb"}}, clusterTime=Timestamp{value=6786856975532556291, seconds=1580188278, inc=3}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBC0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbc"}}, clusterTime=Timestamp{value=6786856975532556292, seconds=1580188278, inc=4}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000052B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBD0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbd"}}, clusterTime=Timestamp{value=6786856975532556293, seconds=1580188278, inc=5}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000062B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBE0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbe"}}, clusterTime=Timestamp{value=6786856975532556294, seconds=1580188278, inc=6}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000072B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBF0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbf"}}, clusterTime=Timestamp{value=6786856975532556295, seconds=1580188278, inc=7}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000082B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC00004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc0"}}, clusterTime=Timestamp{value=6786856975532556296, seconds=1580188278, inc=8}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
==> Let's imagine something wrong happened and I need to restart my Change Stream.
==> Starting from resumeToken={"_data": "825E2FC276000000062B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBE0004"}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000072B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBF0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbf"}}, clusterTime=Timestamp{value=6786856975532556295, seconds=1580188278, inc=7}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000082B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC00004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc0"}}, clusterTime=Timestamp{value=6786856975532556296, seconds=1580188278, inc=8}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000092B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC10004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc1"}}, clusterTime=Timestamp{value=6786856975532556297, seconds=1580188278, inc=9}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000A2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC20004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc2"}}, clusterTime=Timestamp{value=6786856975532556298, seconds=1580188278, inc=10}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000B2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC30004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc3"}}, clusterTime=Timestamp{value=6786856975532556299, seconds=1580188278, inc=11}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000D2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC8F94B5117D894CBB90004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc8f94b5117d894cbb9"}}, clusterTime=Timestamp{value=6786856975532556301, seconds=1580188278, inc=13}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"scores.0.score": 904745.0267635228, "x": 150}}, txnNumber=null, lsid=null}
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000F2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBA0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbba"}}, clusterTime=Timestamp{value=6786856975532556303, seconds=1580188278, inc=15}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"scores.0.score": 2126144.0353088505, "x": 150}}, txnNumber=null, lsid=null}

Como você pode ver aqui, consegui parar de ler meu Change Stream e, a partir do resumeToken que coletei anteriormente, posso iniciar um novo Change Stream a partir deste ponto no tempo.

Código final

ChangeStreams.java (código: ):

Trecho de código

package com.mongodb.quickstart;

import com.mongodb.ConnectionString;
import com.mongodb.MongoClientSettings;
import com.mongodb.client.*;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.quickstart.models.Grade;
import org.bson.BsonDocument;
import org.bson.codecs.configuration.CodecRegistry;
import org.bson.codecs.pojo.PojoCodecProvider;
import org.bson.conversions.Bson;

import java.util.List;
import java.util.function.Consumer;

import static com.mongodb.client.model.Aggregates.match;
import static com.mongodb.client.model.Filters.eq;
import static com.mongodb.client.model.Filters.in;
import static com.mongodb.client.model.changestream.FullDocument.UPDATE_LOOKUP;
import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;

public class ChangeStreams {

public static void main(String[] args) {
        ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
        CodecRegistry pojoCodecRegistry = fromProviders(PojoCodecProvider.builder().automatic(true).build());
        CodecRegistry codecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(), pojoCodecRegistry);
        MongoClientSettings clientSettings = MongoClientSettings.builder()
                                                                .applyConnectionString(connectionString)
                                                                .codecRegistry(codecRegistry)
                                                                .build();

try (MongoClient mongoClient = MongoClients.create(clientSettings)) {
            MongoDatabase db = mongoClient.getDatabase("sample_training");
            MongoCollection<Grade> grades = db.getCollection("grades", Grade.class);
            List<Bson> pipeline;

// Only uncomment one example at a time. Follow instructions for each individually then kill all remaining processes.

/** => Example 1: print all the write operations.
             *  => Start "ChangeStreams" then "MappingPOJOs" to see some change events.
             */
            grades.watch().forEach(printEvent());

/** => Example 2: print only insert and delete operations.
             *  => Start "ChangeStreams" then "MappingPOJOs" to see some change events.
             */
//             pipeline = List.of(match(in("operationType", List.of("insert", "delete"))));
//             grades.watch(pipeline).forEach(printEvent());

/** => Example 3: print only updates without fullDocument.
             *  => Start "ChangeStreams" then "Update" to see some change events (start "Create" before if not done earlier).
             */
//             pipeline = List.of(match(eq("operationType", "update")));
//             grades.watch(pipeline).forEach(printEvent());

/** => Example 4: print only updates with fullDocument.
             *  => Start "ChangeStreams" then "Update" to see some change events.
             */
//             pipeline = List.of(match(eq("operationType", "update")));
//             grades.watch(pipeline).fullDocument(UPDATE_LOOKUP).forEach(printEvent());

/**
             * => Example 5: iterating using a cursor and a while loop + remembering a resumeToken then restart the Change Streams.
             * => Start "ChangeStreams" then "Update" to see some change events.
             */
//             exampleWithResumeToken(grades);
        }
    }

private static void exampleWithResumeToken(MongoCollection<Grade> grades) {
        List<Bson> pipeline = List.of(match(eq("operationType", "update")));
        ChangeStreamIterable<Grade> changeStream = grades.watch(pipeline);
        MongoChangeStreamCursor<ChangeStreamDocument<Grade>> cursor = changeStream.cursor();
        System.out.println("==> Going through the stream a first time & record a resumeToken");
        int indexOfOperationToRestartFrom = 5;
        int indexOfIncident = 8;
        int counter = 0;
        BsonDocument resumeToken = null;
        while (cursor.hasNext() && counter != indexOfIncident) {
            ChangeStreamDocument<Grade> event = cursor.next();
            if (indexOfOperationToRestartFrom == counter) {
                resumeToken = event.getResumeToken();
            }
            System.out.println(event);
            counter++;
        }
        System.out.println("==> Let's imagine something wrong happened and I need to restart my Change Stream.");
        System.out.println("==> Starting from resumeToken=" + resumeToken);
        assert resumeToken != null;
        grades.watch(pipeline).resumeAfter(resumeToken).forEach(printEvent());
    }

private static Consumer<ChangeStreamDocument<Grade>> printEvent() {
        return System.out::println;
    }
}

Lembre-se de descomentar apenas um exemplo de Change Stream por vez.

Encerrando

Os Change Streams são muito fáceis de usar e configurar no MongoDB. Eles são a chave para qualquer sistema de processamento em tempo real.

O único problema restante aqui é como colocar isso em produção corretamente. Change Streams são principalmente um loop infinito, processando um fluxo infinito de eventos. O multiprocessamento é, obviamente, obrigatório para esse tipo de configuração, especialmente se o tempo de processamento for maior do que o tempo que separa 2 eventos.

Se você quiser aprender mais e afundar seu conhecimento com mais rapidez, recomendamos que você confira o treinamento M220J: MongoDB para programadores Java disponível gratuitamente na MongoDB University.

Na próxima publicação no blog, mostrarei as transações ACID multidocumento em Java.

Java - Change Streams

Atualizações

28 de fevereiro de 2024

14 de novembro de 2023

25 de março de 2021

21 de outubro de 2020

Introdução

Configuração

Fluxos de alterações

Um simples Change Streams sem filtros

Um simples Change Stream filtrando pelo tipo de operação

Alterar o comportamento padrão do Stream com operações de atualização

Fluxo de alterações com "Pesquisa de atualização"

Change Streams são retomáveis

Código final

Encerrando

Relacionado

Crie um rastreador de livros Java do MongoDB Spring Boot para iniciantes

Projetos de coleção única no MongoDB com dados Spring (Parte 2)

Driver Java: migrando do 4.11 a 5.0

Introdução ao Azure Spring Apps e ao MongoDB Atlas: um guia passo a passo

Sumário