Akka Persistence Plugin para MongoDB
Akka Persistence viene con tres complementos desarrollados por Lightbend para proporcionar un backend de almacenamiento para actores persistentes:
-
Apache Cassandra
-
JDBC (para RDBMS)
-
Couchbase
Otros backends de almacenamiento están disponibles como complementos comunitarios que se pueden encontrar aquí.
Este es un complemento que implementa MongoDB como un backend de almacenamiento para Akka Persistence. Los backends de almacenamiento típicamente deben implementar las siguientes tres funcionalidades:
-
Escriba en el diario: Persistiendo events por un actor
-
Leer el diario: Proporcionando una vista del lado de lectura CQRS de un actor persistente
-
Almacén de instantáneas: Creación y recuperación de un agregado de todos events hasta cierto punto para un actor
Usando Akka Persistence Plugin para MongoDB
Aquí está cómo comenzar a utilizar MongoDB como un backend de almacenamiento para Akka Persistence utilizando nuestro complemento:
1. Declare una Dependencia
Está disponible en:
<dependency>
<groupId>com.iconsolutions.ipf.core.persistence</groupId>
<artifactId>akka-persistence-mongodb-icon-plugin</artifactId>
</dependency>2. Establecer Akka Persistence Plugin para MongoDB como el Proveedor
Como se indicó anteriormente, el Akka Persistence Plugin para MongoDB implementa cuatro piezas de funcionalidad:
-
escriba un diario
-
leer el diario
-
almacenamiento de instantáneas
-
duradero state
Para configurar Akka Persistence Plugin para MongoDB como el proveedor para escribir, instantánea y duradera state, configure lo siguiente:
akka {
persistence.journal.plugin = "iconsolutions.akka.persistence.mongodb"
persistence.snapshot-store.plugin = "iconsolutions.akka.persistence.mongodb.snapshot"
persistence.state.plugin = "iconsolutions.akka.persistence.mongodb.durable-state"
}3. Configure MongoDB
El complemento utiliza un único mongodb.github.io/mongo-java-driver-reactivestreams/1. 13/javadoc/com/mongodb/reactivestreams/client/MongoClient.html[MongoClient] para conectarse a MongoDB.
Puede especificar la configuración de la conexión utilizando HOCON.
Las propiedades de configuración con sus valores predeterminados se muestran a continuación:
| Clave de configuración | Descripción | Valor predeterminado |
|---|---|---|
|
El MongoDB URI que debe utilizar para conectarse a la base de datos. Establezca el valor de este campo si no está utilizando el complemento desde dentro de IPF o si no tiene la intención de utilizar la misma base de datos que IPF. Honra el global |
|
|
El nombre de la colección que contendrá el domain events. |
|
|
El nombre de la colección que contendrá las instantáneas del actor state. |
|
|
El nombre de la colección que contendrá el actor state. |
|
|
Ya sea que eludir el MongoDB reglas de validación de esquema. |
|
|
El complemento admite dos flavours—MongoDB( Honra el global |
|
|
Debería Desactive esto si al usuario de la base de datos no se le concede el Honra el IPF global |
|
|
El quórum de confirmación para utilizar al crear los índices para Akka Persistence MongoDB. Honra el IPF global |
|
|
Si debe habilitar el soporte SSL. Estableciendo esto a Honra el global |
|
|
Tipo de la tienda de claves. Honra el global |
|
|
Ruta al almacén de claves que contiene el certificado SSL (típicamente un archivo jks). Honra el global |
|
|
Contraseña utilizada para acceder al almacén de claves. Honra el global |
|
|
Contraseña utilizada para acceder a la clave en el almacén de claves. Honra el global |
|
|
Tipo de la tienda de confianza. Honra lo global |
|
|
Ruta al almacén de confianza que contiene el certificado SSL. Honra lo global |
|
|
Contraseña utilizada para acceder al almacén de confianza. Honra el global |
|
|
El número máximo de veces que un failed se intentará la operación (incluido el intento inicial). Honra el global |
|
|
Una lista de MongoDB códigos de error que deben ser reintentados. Honra el global |
|
|
El retraso a utilizar entre cada uno de los intentos si el error en sí no indica una duración de retroceso. Honra el global |
|
Ver Opciones de Cadena de Conexión para todas las opciones que se pueden especificar utilizando el MongoDB Cadena de conexión.
4. Purga de TTL habilitada (Opcional)
Si es necesario, se puede configurar la purga del tiempo de vida (TTL) para su base de datos. Vea el Purgado de Registros y Capturas documentación para más detalles.
Uso del Diario de Lectura
Utilice el diario de lectura para suscribirse al domain events.
Normalmente se utiliza para crear una vista "del lado de lectura" de CQRS basada en el events persistido hasta ahora por un actor persistente.
Las propiedades de configuración con sus valores predeterminados se muestran a continuación:
| Clave de configuración | Descripción | Valor predeterminado |
|---|---|---|
|
Determina si el Flujos de Cambio(MongoDB La implementación de CDC) del diario de lectura se utiliza en lugar de la basada en consultas periódicas. |
|
|
Un período de búfer para ayudar a prevenir la pérdida de datos causada por retrasos de tiempo o condiciones de carrera al recuperar.events. Solo events dentro del [largest_seen_object_id, (current_object_id-allowed_time_drift)] el rango será devuelto por el event consulta de flujo. |
|
|
Define cuánto dato MongoDB debe devolver en un solo lote.
Valores más grandes reducen el número de |
|
El allowed-time-drift ajuste de la event lea la ventana hacia atrás (por la duración proporcionada) para acomodar las diferencias en los relojes del sistema de nodos y posibles retrasos en la replicación.
Esto es esencial en entornos de múltiples nodos, ya que MongoDB’s ObjectId utiliza el tiempo epoch como un desplazamiento, y un nodo lento o desincronizado podría resultar en events no siendo consumidos si se persistieron cerca del límite de la consulta anterior.
Configuración allowed-time-drift a un valor superior al máximo observado akka_event_persistence_time_ns métrica asegura que cualquier events que están retrasados en ser completamente persistidos (debido a la latencia de procesamiento o replicación) todavía se incluyen en la siguiente ventana de lectura. Esto ayuda a prevenir condiciones de carrera y asegura que no events se pierden, incluso si hay un retraso en su disponibilidad.
Si use-change-streams está configurado para true, el complemento cambiará de un enfoque de sondeo, donde las consultas se ejecutan repetidamente para obtener la última events a un Flujos de Cambio enfoque basado, donde los cambios en la colección de diarios son suscritos y cambian events se reciben en tiempo real a través de un cursor de flujo de cambios.
Las propiedades de configuración adicionales específicas para implementaciones de CosmosDB basadas en RU están disponibles cuando use-change-streams está habilitado. Estas propiedades y sus valores predeterminados se muestran a continuación:
| Clave de configuración | Descripción | Valor predeterminado |
|---|---|---|
|
La duración inicial de retroceso antes de restarting el |
|
|
La duración máxima de retroceso antes de restarting el |
|
|
Un factor aleatorio que añade variabilidad a la duración del retroceso. Un factor aleatorio de cero desactiva la variabilidad. |
|
| El enfoque de Change Streams es experimental. Si se utiliza, no se recomienda alternar entre Change Streams e implementaciones basadas en consultas. |
El enfoque de Change Streams puede ser menos intensivo en el MongoDB servidor y lograr un mejor rendimiento, pero es más probable que haya duplicados. Si el complemento no puede mantenerse al día con la tasa a la que events se producen, el cursor puede sobrescribir un event que aún no ha sido gestionado. Cuando esto sucede, el complemento recurrirá a consultas para ponerse al día. Una vez que se haya puesto al día, volverá a utilizar Change Streams.
Como otros Akka Persistence plugins, este plugin proporciona un Scala y Java Variante de DSL para interactuar con el lado de lectura. Puede recuperarlos de la siguiente manera:
// Java DSL
IpfMongoReadJournal javaDslReadJournal = PersistenceQuery
.get(actorSystem)
.getReadJournalFor(IpfMongoReadJournal.class, IpfMongoReadJournal.identifier());
// Scala DSL (but you will probably use this from Scala not Java!)
ReadJournal scalaDslReadJournal = PersistenceQuery
.get(actorSystem)
.readJournalFor(IpfMongoReadJournal.identifier());Estos devolverán un ReadJournal (API de Java, API de Scala) que puede utilizar para ejecutar tres tipos de consultas:
-
eventsByTag: Devuelve un flujo de todos events por etiqueta, dado un desplazamiento. Siuse-change-streamsestá habilitado, transmite continuamente events en tiempo real utilizando Change Streams; de lo contrario, consulta el diario para events desde el desplazamiento hasta la marca de tiempo actual y completa. UtiliceOffset.noOffset()para transmitir desde el principio. -
currentEventsByTag: Siempre consulta el diario (independientemente deuse-change-streamsajustes) y se completa una vez que todos events hasta la marca de tiempo actual han sido consumidos. -
currentEventsByPersistenceId: Devuelve un flujo de todos events por persistenceId, dado el rango de números de secuencia proporcionado (inclusive).
| Todas las consultas anteriores tienen un tipo de retorno de:`Source<EventEnvelope, NotUsed>` |
Consejos para Utilizar el Diario de Lectura
Aquí hay algunos consejos que pueden ser útiles al escribir un diario de lectura utilizando el Akka Persistence Plugin para MongoDB:
-
Es responsabilidad del consumidor almacenar el desplazamiento que fue proporcionado por última vez por el complemento.
-
Envuelva el
eventsByTagoperaciones en un RestartSource para que cuando/si la transmisión falla debido a algún error, pueda automáticamente restart desde el último desplazamiento que fue persistido, en lugar de requerir un restart de la aplicación -
Se recomienda guardar los desplazamientos por lotes para reducir la sobrecarga de rendimiento.
-
Es, por supuesto, posible persistir el desplazamiento que su aplicación vio por última vez en otro MongoDB colección
-
Si la aplicación del consumidor se cierra inesperadamente después de algún events se han procesado, pero antes de que se guardara el desplazamiento, el complemento enviará duplicados.events entre el último desplazamiento guardado y el fallo. La aplicación de downstream debe ser capaz de manejar tales duplicados.
Girando MongoDB Desactivación de la Validación de Documentos para Asegurar un Alto Rendimiento
Por defecto,MongoDB habilita estricto validación de esquema para todos los documentos insertados o actualizados.
A pesar de que la colección utilizada para el event el diario no requiere (ni incluye) ninguna validación de esquema, se aconseja desactivar explícitamente la validación, lo cual se puede lograr especificando un validationLevel: 'off' opción al crear la colección, p. ej.
db.createCollection("journal",
{
validationLevel: "off"
})Modo de Base de Datos
El complemento es compatible con otras bases de datos que implementan el MongoDB protocolo de red, como Azure CosmosDB por Microsoft, o AWS DocumentDB.
Algunas de estas bases de datos tienen limitaciones en el tipo de índices que se pueden crear, específicamente en cuanto a si la base de datos admite la creación de campos compuestos que presenten un campo anidado.
El complemento crea un índice para el eventsByTag consulta como parte de la Consulta de Persistencia.
Si utiliza el complemento con un MongoDB- base de datos compatible (no MongoDB en sí mismo), y la base de datos de destino no admite campos anidados como parte de índices compuestos, establezca el iconsolutions.akka.persistence.mongodb.database-mode
clave de configuración a algo diferente de mongo.
Esto creará un índice ligeramente menos eficiente, pero que es compatible con dichas bases de datos de terceros.
Leer preocupación
Las colecciones de diario y de instantáneas pueden necesitar una preocupación de lectura más confiable que la que se utiliza como predeterminada en la base de datos, en cuyo caso el readConcernLevel opción de cadena de conexión debe ser utilizado para configurarlo.
Ver los documentos oficiales de consistencia causal para obtener detalles sobre por qué esta configuración es importante, y este enlace para obtener más detalles sobre cada una de las opciones de preocupación de lectura disponibles.
Para cómo los ajustes de preocupación de lectura y escritura se relacionan con CosmosDB. MongoDB API, por favor consulte el documentación oficial de Azure.
Purgado de Registros y Capturas
La purga de la base de datos puede ser habilitada para ayudar a garantizar un rendimiento consistente de la base de datos y evitar el riesgo de que una base de datos se vuelva inaccesible debido a llenar todo el almacenamiento disponible. Para las colecciones de diarios y instantáneas, la purga puede ser gestionada utilizando la funcionalidad de Tiempo de vida (TTL). La funcionalidad TTL permite que la base de datos expire automáticamente los datos de las colecciones de diario y de instantáneas.
| Al construir un dominio utilizando IPF Flo-Lang, el Akka EventSourcedBehaviour comandos el Akka Persistence Complemento para crear instantáneas y eliminar el diario events. Hay una serie de opciones configurables para determinar cuándo se deben crear las instantáneas y si events debe ser eliminado, más información sobre esta configuración se puede encontrar en el Documentación de purga de Flo-Lang. Esto debe configurarse junto con los pasos a continuación para purgar datos con éxito. |
Para habilitar la purga de TTL, se deben seguir pasos para configurar la base de datos utilizada y, si se está utilizando CosmosDB, configurar el Akka Persistence MongoDB Complemento.
MongoDB configuración
Al utilizar MongoDB, el índice TTL es un índice especial de un solo campo configurado para un campo de fecha BSON. Para más información, consulte el oficial MongoDB Documentación de TTL.
Índice de colección de revistas
Para la colección de revistas, se debe crear un índice TTL para el deletedAt campo.
Un ejemplo MongoDB el comando de shell para crear este índice es:
db.getCollection("journal").createIndex( { "deletedAt": 1 }, {
expireAfterSeconds: 31536000 } )Este comando significa que las entradas del diario son elegibles para ser eliminadas de la base de datos 31536000 segundos (365 días) después del valor de la fecha de la deletedAt campo.
El complemento Write Journal es responsable de establecer el deletedAt campo al marcar el Diario Events para eliminación.
Índice de Colección de Instantáneas
Para la colección de instantáneas, se debe crear un índice TTL para el campo 'insertedAt'. Un ejemplo MongoDB el comando de shell para crear este índice es:
db.getCollection("snapshots").createIndex( { "insertedAt": 1 }, {
expireAfterSeconds: 31536000 } )Este comando significa que las entradas de instantáneas son elegibles para su eliminación de la base de datos 31536000 segundos (365 días) después del valor de la fecha de la insertedAt campo
Migrando Datos Existentes
Documentos existentes en el journal y snapshots Las colecciones no contienen un campo BSON Date, por lo que, por defecto, los índices de la colección no tendrán efecto en los datos existentes.
Se recomienda actualizar los datos existentes dentro de estas colecciones para establecer un deletedAt y insertedAt valor para el Diario Events y Snapshots, respectivamente.
Para salvaguardar contra la eliminación involuntaria de datos no terminales (ya que usted puede no saber si los datos existentes han alcanzado un estado terminal state o no), puede actualizar los campos TTL respectivos para tener una fecha BSON en el futuro.
Por ejemplo, si el índice configurado expireAfterSeconds evalúa a 20 días, y usted establece un Diario Event’s deletedAt el campo a 10 días en el futuro, el documento será purgado 30 días después de que se realice la actualización.
Vea el Documentación de migración de purga para obtener más información sobre la configuración y el sistema IPF existente para la purga de Journal y Snapshot.
Cosas a considerar
MongoDB La funcionalidad TTL requiere que se establezca un campo de fecha BSON, por lo que si ese campo no existe en el documento, no será [.underline]#eliminado#. Por lo tanto, si un Diario Event no está marcado para ser eliminado, no será purgado por el MongoDB Índice TTL.
Como se especifica en el Documentación de TTL de MongoDB Eliminar un gran número de documentos a la vez crea una carga de trabajo considerable y puede causar problemas de rendimiento. Esto es principalmente algo a considerar para la migración de datos existentes, y se recomienda que no intente purgar todos los datos existentes de una sola vez. En su lugar, recomendamos que la purga de los datos existentes se realice durante horas no laborables o en lotes.
Por ejemplo, si el índice configurado expireAfterSeconds evalúa a 20 días, usted puede querer actualizar todos los Journal existentes Events tener un deletedAt valor de 2024-07-17T03:00:00. 000Z.
En este escenario, el documento sería scheduled será eliminado a las 3 a.m. del 27/07/2024.
Configuración de CosmosDB
Al utilizar índices TTL con CosmosDB, es necesario habilitar la configuración para el Akka Persistence Plugin y para cada colección de CosmosDB.
Existen dos tipos de índices TTL de CosmosDB soportados: establecer un valor TTL predeterminado en toda la colección y establecer valores TTL individuales para cada documento. Para más información, consulte la documentación oficial de TTL de CosmosDB.
El índice de nivel de colección está configurado en el específico de CosmosDB._ts campo.
Una vez que se crea el índice, la base de datos eliminará automáticamente cualquier documento en esa colección que no haya sido modificado durante una cantidad configurada de segundos.
El nivel de TTL del documento no está configurado como un índice, sino que es un valor que existe en cada documento.
El(los) documento(s) deben contener una propiedad de nivel raíz.ttl.
El ttl el valor debe ser un int32 (o un int64 que se ajuste a un int32), y debe haberse creado un índice TTL a nivel de colección como se describió anteriormente para esa colección.
Los valores de TTL establecidos en un documento anularán el valor de TTL de la colección.
Índice de colección de revistas
Para la colección de revistas, se debe crear un índice TTL a nivel de colección para el _ts campo.
Un ejemplo MongoDB el comando de shell para crear este índice es:
db.getCollection("journal").createIndex( { "_ts": 1 }, {
expireAfterSeconds: 31536000 } )Este comando significa que las entradas del diario son elegibles para ser eliminadas de la base de datos 31536000 segundos (365 días) después de haber sido modificadas por última vez.
Además de esto, la colección de revistas también debe utilizar propiedades TTL a nivel de documento.
Cuando el Akka Persistence el complemento está configurado correctamente, la Entrada del Diario ttl la propiedad se establecerá al eliminar a través de la Akka Persistence Plugin Escribir Diario.
Akka Persistence Configuración del Plugin
Se deben establecer configuraciones adicionales para el Akka Persistence Complemento para habilitar un TTL a nivel de documento para el Diario Events. La siguiente duración basada HOCON debe ser configurado:
iconsolutions.akka.persistence.mongodb.write-journal.cosmos-ttl = 365dEsto permite que el Plugin establezca el ttl propiedad para el Diario Events cuando son eliminados por el Registro de Escritura.
Utilizando el valor de ejemplo anterior, Diario Events se actualizan para establecer el ttl propiedad a 31536000 (365 días en segundos).
Esto habilita la propiedad TTL a nivel de documento para CosmosDB que anulará el índice TTL a nivel de colección.
Si el iconsolutions.akka.persistence.mongodb.write-journal.cosmos-ttl la configuración no se anula explícitamente, entonces el nivel del documento ttl la propiedad no se establecerá en el Diario Event eliminación.
Si desea utilizar la purga de CosmosDB, se recomienda que anule esta configuración.
|
Índice de Colección de Instantáneas
Para la colección de instantáneas, se debe crear un índice TTL a nivel de colección para el _ts campo.
Un ejemplo MongoDB el comando de shell para crear este índice es:
db.getCollection("snapshots").createIndex( { "_ts": 1 }, {
expireAfterSeconds: 31536000 } )Este comando significa que las entradas de instantáneas son elegibles para ser eliminadas de la base de datos 31536000 segundos (365 días) después de haber sido modificadas por última vez.
Migrando Datos Existentes
Por defecto, CosmosDB automáticamente llenará y actualizará el _ts campo para un documento dado, manteniendo la marca de tiempo de la última modificación del documento.
Creando un índice a nivel de colección para ambos el journal y snapshots la colección tendrá el mismo efecto que se ha descrito anteriormente.
Puede que desee actualizar el Diario existente. Events tener un valor TTL a nivel de documento. Esto sobrescribirá el valor del índice TTL de la colección y deberá configurarse manualmente. Vea el documentación oficial de CosmosDB para obtener más información sobre cómo establecer un valor TTL a nivel de documento.
Como ejemplo, puede que desee establecer el journal índice a nivel de colección de 365 días, pero usted desea todos los Journal existentes Events será eliminado en 30 días.
Así que usted actualizaría todos los diarios existentes. Events para establecer el ttl campo a 2592000 (30 días en segundos).
Vea el Documentación de migración de purga para obtener más información sobre la configuración y el sistema IPF existente para la purga de Journal y Snapshot.
Cosas a considerar
Desafortunadamente, debido a que CosmosDB requiere que se configure un índice TTL a nivel de colección, existe la posibilidad de que algunos Journal Events podría ser eliminado antes de que se cree un Snapshot para el ID de Persistencia dado. Dependiendo del valor TTL que haya configurado, esto es poco probable que ocurra, pero sigue siendo una posibilidad de la que debe estar consciente.
En CosmosDB, la eliminación de elementos expirados es una tarea en segundo plano que consume Unidades de Solicitud sobrantes; si no hay suficientes Uds. de Solicitud disponibles, la eliminación de datos se retrasa. Por lo tanto, no necesita tener en cuenta el mismo impacto en el rendimiento de eliminar muchos documentos a la vez como podría necesitar al utilizar Purgado basado en TTL de MongoDB