Overview
En esta guía, puedes aprender a usar un flujo de cambios para supervisar los cambios en tiempo real de tus datos. Un flujo de cambios es una funcionalidad de MongoDB Server que permite que la aplicación se suscriba a cambios en los datos de una colección, base de datos o implementación.
Datos de muestra
Los ejemplos de esta guía utilizan el sample_restaurants.restaurants colección de la Conjuntos de datos de muestra de Atlas. Para aprender a crear un clúster gratuito de MongoDB Atlas y cargar los conjuntos de datos de muestra, consulta el
Guía de primeros pasos.
Importante
Proyecto Reactor librería
Esta guía usa la librería Proyecto Reactor para consumir instancias Publisher devueltas por los métodos del driver Reactive Streams de Java. Para obtener más información sobre la biblioteca Project Reactor y cómo utilizarla, consulta Primeros pasos en la documentación de Reactor. Para obtener más información sobre cómo utilizamos los métodos de la librería Project Reactor en esta guía, consulta la guía Guardar datos en MongoDB.
Abre un flujo de cambios
Para abrir un flujo de cambios, llama al método watch(). La instancia sobre la que llama el método determina el alcance de los eventos a los que el flujo de cambios escucha. Puedes llamar al método watch() en instancias de las siguientes clases:
MongoClient: Para supervisar todos los cambios en la implementación de MongoDBMongoDatabase: Para supervisar los cambios en todas las colecciones de la base de datosMongoCollection: Para supervisar los cambios en la colección
El siguiente ejemplo abre un flujo de cambios en la colección restaurants y muestra los cambios a medida que ocurren:
// 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 comenzar a observar los cambios, ejecuta la aplicación. Luego, en una aplicación o shell por separado, realiza una operación de guardar en la colección restaurants. Actualizar un documento que tiene un valor de campo "name" de "Blarney
Castle" da como resultado la siguiente salida de flujo de cambios:
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 la Salida del Cambio de Flujo
Puedes pasar un pipeline de agregación como parámetro al método watch() para modificar la salida del flujo de cambios. Este parámetro te permite supervisar solo eventos de cambio especificados.
Puedes especificar las siguientes etapas de agregación en el parámetro pipeline:
$addFields$match$project$replaceRoot$replaceWith$redact$set$unset
El siguiente ejemplo pasa un pipeline de agregación a un flujo de cambios para registrar solo las operaciones de actualización:
// 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 obtener más información sobre cómo modificar la salida de tu flujo de cambios, consulta la sección Modificar salida del flujo de cambio en el manual del MongoDB Server.
Modificar el comportamiento de watch()
Puedes encadenar métodos al método watch() que representan opciones que puedes usar para configurar la operación del flujo de cambios. Si no especificas ninguna opción, el driver no personaliza la operación.
La siguiente tabla describe los métodos que puedas encadenar a watch() para personalizar su comportamiento:
Opción | Descripción |
|---|---|
| Specifies whether to show the full document after the change, rather
than showing only the changes made to the document. To learn more about
this option, see Include Pre-Images and Post-Images. |
| Specifies whether to show the full document as it was before the change, rather
than showing only the changes made to the document. To learn more about
this option, see Include Pre-Images and Post-Images. |
| Directs watch() to resume returning changes after the
operation specified in the resume token.Each change stream event document includes a resume token as the _id
field. Pass the entire _id field of the change event document that
represents the operation you want to resume after.resumeAfter() is mutually exclusive with startAfter() and startAtOperationTime(). |
| Directs watch() to start a new change stream after the
operation specified in the resume token. Allows notifications to
resume after an invalidate event.Each change stream event document includes a resume token as the _id
field. Pass the entire _id field of the change event document that
represents the operation you want to resume after.startAfter() is mutually exclusive with resumeAfter() and startAtOperationTime(). |
| Directs watch() to return only events that occur after the
specified timestamp.startAtOperationTime() is mutually exclusive with resumeAfter() and startAfter(). |
| Specifies the maximum amount of time, in milliseconds, the server waits for new
data changes to report to the change stream cursor before returning an
empty batch. Defaults to 1000 milliseconds. |
| Starting in MongoDB Server v6.0, change streams support change notifications
for Data Definition Language (DDL) events, such as the createIndexes and dropIndexes events. To
include expanded events in a change stream, call this method and pass
in the value, true. |
| Specifies the maximum number of change events to return in each batch of the
response from the MongoDB cluster. By default, the driver sets this value to
Long.MAX_VALUE.A batchSize of 0 means that the cursor will be established, but no documents
will be returned in the first batch. |
| Specifies the collation to use for the change stream cursor. |
| Attaches a comment to the operation. |
Incluir imágenes preoperativas y postoperatorias
Importante
Puede habilitar pre imágenes y post imágenes en colecciones solo si su implementación utiliza MongoDB v6.0 o posterior.
De forma predeterminada, cuando realizas una operación en una colección, el evento de cambio correspondiente solo incluye el delta de los campos modificados por esa operación. Para ver el documento completo antes o después de un cambio, encadena el método fullDocumentBeforeChange() o fullDocument() al método watch().
La preimagen es la versión completa de un documento antes de un cambio. Para incluir la preimagen en el evento de flujo de cambios, se debe pasar uno de los siguientes valores al método fullDocumentBeforeChange():
FullDocumentBeforeChange.WHEN_AVAILABLE: El evento de cambio incluye una preimagen del documento modificado para los eventos de cambio solo si la preimagen está disponible.FullDocumentBeforeChange.REQUIREDEl evento de cambio incluye una preimagen del documento modificado para los eventos de cambio. Si la preimagen no está disponible, el driver genera un error.
La post-imagen es la versión completa de un documento después de un cambio. Para incluir la post-imagen en el evento del flujo de cambios, pasa uno de los siguientes valores al método fullDocument():
FullDocument.UPDATE_LOOKUP: El evento de cambio incluye una copia de todo el documento cambiado a partir de cierto tiempo después del cambio.FullDocument.WHEN_AVAILABLE: El evento de cambio incluye una postimagen del documento modificado para eventos de cambio solo si la postimagen está disponible.FullDocument.REQUIRED: El evento de cambio incluye una imagen posterior del documento modificado para eventos de cambio. Si la imagen posterior no está disponible, el driver genera un error.
El siguiente ejemplo abre un flujo de cambios en una colección e incluye la imagen posterior de documentos actualizados enlazando el método fullDocument() al 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 obtener más información sobre pre-imágenes y post-imágenes, consulta Change Streams con preimágenes y postimágenes de documentos en el manual de MongoDB Server.
Información Adicional
Para saber más sobre las change streams, consulta Change Streams en el manual de MongoDB Server.
Documentación de la API
Para aprender más sobre cualquiera de los métodos o tipos analizados en esta guía, consulta la siguiente documentación de API: