Transacción Caching

IPF es un event-aplicación de origen y como resultado el lado de lectura ("consulta") es eventually consistente. Esto significa que en los casos en los que necesitamos examinar la historia más reciente de la aplicación, puede que no sea representativo consultar el lado de lectura, ya que puede que aún no se haya actualizado con toda events que han tenido lugar en el lado de escritura.

Como resultado, los usuarios tienen la capacidad de implementar una transacción.cache, utilizando la transacción suministrada cache servicio. Se puede utilizar para satisfacer los requisitos empresariales tales como:

Verificaciones de duplicados funcionales
Verificaciones técnicas de duplicados

Poblando la Transacción Cache

La transacción cache requiere algo de ensamblaje, ¡Pero no es tan complicado comenzar! Aquí está lo que usted necesita hacer:

1. Agregue la dependencia

Si utiliza el Icon BOM, añada la siguiente dependencia a su -app módulo:

<dependency>
    <groupId>com.iconsolutions.ipf.core.platform</groupId>
    <artifactId>ipf-transaction-cache</artifactId>
</dependency>

2. Identifique los tipos de transacción a persistir

Debe identificar lo que desea persistir en el cache. El PersistentTransactionCacheService tiene un tipo genérico T que puede utilizar para insertar cualquier tipo de MongoDB POJO persistente. En nuestro ejemplo, queremos persistir este objeto de pago:

    public static class Payment {
        private final String from;
        private final String to;
        private final BigDecimal amount;
        private final LocalDate paymentDate;
    }

3. Seleccione business data eso es importante

Necesitamos implementar un "extractor de ID" para determinar qué campos son importantes al determinar si hemos visto esta transacción anteriormente. Algunos ejemplos podrían ser:

Cantidad
ID de extremo a extremo
Desde el Identificador
Para Identificador
Datos no estructurados

Esto se implementa como un Function<T, List<String>>.

Esto proporciona al servicio una forma de extraer los campos relevantes y hasharlos juntos para intentar buscarlos de manera eficiente más tarde.

Aquí tiene un ejemplo de cómo inicializar la transacción.cache servicio:

        var transactionCacheService = new PersistentTransactionCacheService<>(
                payment -> List.of(payment.getFrom(), payment.getTo(), payment.getAmount().toString())
                , repo, repositoryRetryProvider);

El primer argumento consiste en la lista de campos (que deben estar protegidos contra nulos!) que forman parte del hash. La implementación particular de cache el servicio que hemos seleccionado aquí es MongoDB-basado, por lo que el segundo argumento toma un repositorio para almacenar cache entradas. Diferentes implementaciones pueden tener diferentes firmas dependiendo de sus requisitos.

4. Cree un enum de tipo de entrada

El cache puede contener potencialmente diferentes tipos de transacciones. Por esa razón, necesitamos poder enumerar los diferentes tipos. Esto está representado por el TransactionCacheEntryType interfaz. Aquí está su definición:

public interface TransactionCacheEntryType {
    String getName();
}

Podemos ver que en realidad es solo una forma de poder diferenciar entre los diferentes tipos de transacciones que están cached. Necesitamos esto porque algunos flujos de transacción pueden compartir la misma raíz.message type(Considere los mensajes entrantes y salientes del mismo tipo e.j.pacs. 008).

Aquí tiene un ejemplo de implementación de un TransactionCacheEntryType:

    public enum ExampleTransactionCacheEntryType implements TransactionCacheEntryType {
        TYPE_ONE,
        TYPE_TWO;

        @Override
        public String getName() {
            return name();
        }
    }

Este es un enum que implementa el TransactionCacheEntryType interfaz y puede soportar dos tipos diferentes de cache entradas:`TYPE_ONE` y TYPE_TWO.

Podemos entonces utilizar el servicio para persistir nuestros tipos para persistir.

5. Envuelva y guarde

Ahora podemos llamar a la transacción cache servicio para guardar nuestro Payment con su tipo así:

        var payment = new Payment("Me", "You", new BigDecimal("4.20"), LocalDate.now());

        var saveFuture = transactionCacheService.saveToCache(TYPE_ONE, payment, "messageId");

Para evitar guardar dos entradas para el mismo mensaje físico (por ejemplo, en caso de un reintento o reactivación), llamamos saveToCache método con un parámetro messageId. El MessageId debe ser un identificador único para el contenido que estamos almacenando en el cache y normalmente sería el persistenceId en un flujo.

El messageId no se refiere al msgId en un Iso20022 mensaje

Una nueva entrada no se guardará si ya existe una entrada con el mismo hash y messageId en el cache para el tipo dado, en su lugar devolverá el registro existente.

Verificando la Transacción Cache

El cache el servicio tiene el siguiente método para recuperar datos de la cache:

    CompletionStage<List<TransactionCacheEntry<T>>> findInCache(TransactionCacheEntryType type, T content);

Necesita el tipo de entrada que desea encontrar, seguido de la T escriba el tipo que desea verificar para ver si es un duplicado funcional.

Devuelve un futuro que contiene una lista de coincidencias.cache entradas. Puede desear inspeccionar su creationDate para verificar duplicados funcionales dentro de un cierto intervalo de tiempo.

Consideraciones de Implementación

Purga (TTL)

Para el MongoDB implementación, considere utilizar un Índice TTL de MongoDB en el creationDate campo para caducar entradas. Se crea un índice para buscar por hash de forma predeterminada, pero puede desear agregar un índice TTL para expirar (eliminar) entradas después de un período de tiempo específico si ya no son necesarias.

La creación de índices predeterminados puede ser desactivada con:

ipf.transaction-cache.mongodb.create-indexes=false

Los índices pueden ser deshabilitados globalmente con:

ipf.mongodb.create-indexes=false

Para deshabilitar la indexación globalmente pero mantenerla para la transacción cache, aplique lo siguiente, manteniendo el orden:

ipf.mongodb.create-indexes=false
ipf.transaction-cache.mongodb.create-indexes=true

Compromiso de Quórum

El quórum de confirmación puede ser controlado de manera similar con:

ipf.transaction-cache.mongodb.commit-quorum=1

O bien anulado globalmente con:

ipf.mongodb.commit-quorum=1

Purga (ProgramadorDePurgaDeCacheDeTransacciones)

Para el MongoDB En la implementación, también existe la opción de programar un trabajo recurrente para eliminar todas las entradas de un tipo y edad específicos utilizando el TransactionCachePurgingScheduler.

El TransactionCachePurgingScheduler está disponible dentro de la Transacción Cache módulo y no es necesario añadir dependencias adicionales.

1. Proporcione su configuración:

your.purging.config.path{
  transaction-cache-entry-type = "TYPE_ONE"
  retain-from-time =  "17:00:00"
  retain-from-offset = "1 day"
  scheduling-specification =  "0 0 17 ? * *"
}

transaction-cache-entry-type: Debe coincidir con la cadena proporcionada por el método getName() de su TransactionCacheEntryType.
retain-from-time: La hora en el día de ejecución de la purga de la que desea retener entradas.
retain-from-offset: Proporcione una duración que se restará del tiempo de retención. Debe ser un duración de hocon. Utilice 0 días si no se necesita compensar desde el tiempo de retención.
scheduling-specification: Una cadena que representa un cron expression de cuándo ejecutar el trabajo de purga. Para obtener ayuda con la construcción de un cron expression, utilice un en línea cron expression constructor como este.

El ejemplo de trabajo anterior se ejecutará a las 5 p.m. todos los días."0 0 17? * *") y eliminará todo TYPE_ONE entradas de su transacción cache que son anteriores a las 5 p.m. del día anterior a la ejecución del trabajo ("17:00:00" menos el desplazamiento de "1 day"). Por ejemplo, si se ejecuta a las 17:00 del 23 de abril de 2024, se eliminarían las entradas anteriores a las 17:00 del 22 de abril de 2024.

Tenga cuidado en la relación entre su scheduling specification y su retención de tiempo. Es posible que el trabajo se ejecute antes de su tiempo de retención.

Ejecutar un trabajo de purga podría resultar en la eliminación de un número muy grande de entradas calificadas a la vez. Esta gran carga de trabajo puede causar problemas de rendimiento. Considere scheduling trabajos para horas fuera de servicio/períodos de tranquilidad para reducir este riesgo.

2. Cree un bean

Defina un bean para el TransactionCachePurgingScheduler dentro de una clase de configuración de Spring relevante o de autoconfiguración.

@Configuration
public class TransactionCachePurgingConfig {

    @Bean
    public TransactionCachePurgingScheduler<Payment> transactionCachePurgingScheduler(
            TransactionCacheService<Payment> transactionCacheService,
            SchedulingModuleInterface schedulingModule,
            Clock clock,
            Config config) {
        return new TransactionCachePurgingScheduler<>(
                transactionCacheService,
                schedulingModule,
                clock,
                config.getConfig("your.purging.config.path"));
    }
}

Sustituir <Payment> para el nombre de clase relevante a su transacción cache.

Enfoque de purga a utilizar

El enfoque de purga que usted querrá utilizar dependerá de su caso de uso. Las diferencias entre los enfoques y la recomendación para su uso se resumen en la tabla a continuación.

Tipo	Acción	Uso recomendado
TTL	Elimina todos los datos de la colección en función de su antigüedad en la colección.	Eliminación de todos los datos a una edad específica donde no tiene la necesidad de ser selectivo.
TransactionCachePurgingScheduler	Elimina datos por tipo y antigüedad (en un momento especificado)	Eliminación de conjuntos de datos a diferentes frecuencias y controlando cuándo ocurre la purga

Tipo

Acción

Uso recomendado

TTL

Elimina todos los datos de la colección en función de su antigüedad en la colección.

Eliminación de todos los datos a una edad específica donde no tiene la necesidad de ser selectivo.

TransactionCachePurgingScheduler

Elimina datos por tipo y antigüedad (en un momento especificado)

Eliminación de conjuntos de datos a diferentes frecuencias y controlando cuándo ocurre la purga