Cache Strategies with Redisson & Exposed R2dbc

This document explains how to implement various caching strategies and provides examples, utilizing the Redisson and R2dbc support of Exposed 1.0.0-beta-2.

Transcript

1. Debop (sunghyouk.bae@gmail.com) 2025-05-29 Cache Strategies - R2dbc With Redisson, Exposed R2dbc
2. Agenda • ׮নೠ நद ੹ۚ Cache Strategies • Cache Aside, Read Through, Write Through, Write Behind • Cache with Redisson • MapLoaderAsync, MapWriteAsync • RMapCache, RLocalCachedMap • Interactive with DB by Exposed R2dbc
3. Cache Strategies - R2dbc
4. Cache Aside ಁఢ গ೒ܻா੉࣌੉ ݢ੷ நदܳ ઑഥೞҊ, நदী ؘ੉ఠо হਵݶ DBীࢲ ੍যৡ ٍ நदী ੷੢ೠ׮. ߈ࠂ੸ੋ ੍ӝо ݆਷ ؘ੉ఠী ੸೤ 1. Read Cache asynchronous 2. Read from DB with R2dbc 3. Write Async to Cache
5. Read Through ಁఢ நदী ؘ੉ఠо হਸ ٸ நदо ૒੽ DBীࢲ ؘ੉ఠܳ ઑഥ೧ நदী ੷੢ೠ׮. গ೒ܻா੉࣌਷ ೦࢚ நद݅ ੽Ӕೠ׮. 1. Read from Cache 2. Read from DB with R2dbc 3. Read from Cache
6. Write Through ಁఢ ؘ੉ఠܳ நद৬ DBী زदী ੷੢ೠ׮. ؘ੉ఠ ੌҙࢿ੉ ࠁ੢غ૑݅, ॳӝ ࢿמ੉ ੷ೞؼ ࣻ ੓׮ 1. Write to Cache 2. Write to DB with R2dbc
7. Write Behind ؘ੉ఠܳ ਋ࢶ நदী ੷੢ೞҊ, ੌ੿ ઱ӝ۽ நदীࢲ DB۽ ੌҚ ੷੢೤פ׮. ॳӝ ࢿמ੉ જ૑݅, நद ੢গ द ؘ੉ఠ ਬप ਤ೷੉ ੓׮ 1. Write Multiple Items to Cache 2. Write to DB with R2dbc at once
8. Write Around ؘ੉ఠܳ DBী݅ ੷੢ೞҊ, நदח јनೞ૑ ঋח׮. ੉റ ઑഥ द நदী ੸੤ؾפ׮. ॳӝо ࠼ߣೞ૑݅ ੍ӝח ٘ޙ ҃਋ী ੸೤ 4. Read from Cache 2. Read from DB with R2dbc 3. Write to Cache 1. Write to DB with R2dbc
9. நद ੹ۚ߹ ੸ਊ ߑউ ੹ۚݺ ബҗ੸ੋ ࢚ട ੢੼ ױ੼ ߂ ઱੄੼ Cache-Aside ੍ӝ ੘স੉ ݆Ҋ, ؘ੉ఠ ߸҃੉ ੸਷ ҃਋ ߧਊ੸, நद ੢গী ఍۱੸, ҳഅ੉ рױ ؘ੉ఠ ୭नࢿ ҙܻ ೙ਃ Read-Through زੌ ؘ੉ఠо ߈ࠂ੸ਵ۽ ੍൤ח ҃਋ ੍ӝ ࢿמ ӓ؀ച, ੌҙࢿ ਬ૑ ਊ੉ ୐ ਃ୒ द ૑ো, ਕ߁স ೙ਃ Write-Through ؘ੉ఠ ੌҙࢿ੉ ݒ਋ ઺ਃೠ ҃਋ ؘ੉ఠ ੌҙࢿ ࠁ੢ ॳӝ ૑ো ߊࢤ Write-Back ॳӝ ੘স੉ ݆Ҋ, ੌद੸ ؘ੉ఠ ਬपਸ хࣻೡ ࣻ ੓ח ҃਋ ॳӝ ࢿמ ӓ؀ച, DB ࠗೞ хࣗ நद ੢গ द ؘ੉ఠ ਬप ਤ೷ Write-Around ؘ੉ఠо ੗઱ ॳ੉૑݅, ੗઱ ੍൤૑ ঋח ҃਋ ࠛ೙ਃೠ நद য়৏ ߑ૑ ੍ӝ ࢿמ਷ Cache-Asideࠁ׮ ծਸ ࣻ ੓਺
10. Local Cache Application ࢲߡী ੷੢, ࢲߡ߹ ࠛੌ஖ ߊࢤ ਤ೷ 1. Read from DB with R2dbc 2. Update to DB with R2dbc 3. Read from DB with R2dbc
11. Global Cache Redis, Memcached ١ ߹ب੄ ઺ঔ நद ࢲߡܳ ࢎਊ, ֎౟ਖ ࠗೞ ࢚ઓ
12. Near Cache Local Cache + Global Cache ഋక੄ ೞ੉࠳ܻ٘ ҳઑ 1. Sync cache 2. Update Cache 2.1 Sync updated cache items 2.1 Sync updated cache items
13. நद ੷੢ࣗ ਬഋ Local Cache vs Global Cache vs Near Cache ҳ࠙ Local Cache (۽ஸ நद) Global Cache (Ӗ۽ߥ நद) Near Cache (פয நद) ҳࢿ п ࢲߡ(ੋझఢझ) ղࠗী நदܳ م ߹ب੄ ઺ঔ நद ࢲߡ(৘: Redis, Memcached) ࢎਊ ௿ۄ੉঱౟ ژח WAS খױী ۽ஸ நद ୶о ࣘب ݒ਋ ࡅܴ (֎౟ਕ௼ ࠛ೙ਃ, ݫݽܻ ੽Ӕ) ֎౟ਕ௼ ҃ਬ۽ Local Cacheࠁ׮ וܿ Local Cache ࣻળ੄ ࡅܲ ࣘب ੌҙࢿ ࢲߡ р ؘ੉ఠ ࠛੌ஖ оמ, زӝച য۰਑ ࢲߡ р ؘ੉ఠ ੌҙࢿ ࠁ੢ Ӗ۽ߥ நद৬ زӝച ೙ਃ ഛ੢ࢿ ࢲߡ ࣻ טܾࣻ۾ நद ઺ࠂ, ݫݽܻ ࠺ബਯ ࢲߡ ࣻ ט۰ب ؘ੉ఠ ઺ࠂ ੸਺, ഛ੢ࢿ ਋ࣻ ࢲߡ ࣻ ૐо द Ӗ۽ߥ நद ࠗೞ ࠙࢑ ੢੼ - ୡҊࣘ ੽Ӕ - ֎౟ਕ௼ ੢গ ৔ೱ হ਺ - ؘ੉ఠ ҕਬ ਊ੉ - ੌҙࢿ ਬ૑ - ഛ੢ࢿ જ਺ - ࡅܲ ੽Ӕ - Ӗ۽ߥ நद ࠗೞ хࣗ ױ੼ - ؘ੉ఠ ੌҙࢿ ҙܻ য۰਑ - ݫݽܻ խ࠺ - ֎౟ਕ௼ ૑ো - ױੌ ੢গ੼ оמࢿ - زӝച ࠂ੟ - ୶о ݫݽܻ ೙ਃ ੸೤ ࢎ۹ ߸҃ ੸Ҋ ࢲߡ߹ ة݀੸ ؘ੉ఠ ৈ۞ ࢲߡо زੌ ؘ੉ఠ ҕਬ, ੌҙࢿ ઺ਃೠ ҃਋ ؀ӏݽ ࠙࢑ ജ҃ীࢲ நद ബਯ ӓ؀ച
14. Implementation Cache Strategies with Redisson & Exposed R2dbc
15. Redisson RMap<K,V> MapLoaderAsync, MapWriterAsync MapLoaderAsync<K,V> MapWriterAsync<K,V> RMap<K,V> RMapCache<K,V> RLocalCachedMap<K,V>
16. MapLoaderAsync<ID, E> open class R2dbcEntityMapLoader<ID: Any, E: HasIdentifier<ID>>( private val loadByIdFromDB: suspend (ID) -> E?, private val loadAllIdsFromDB: suspend (channel: Channel<ID>) -> Unit, private val scope: CoroutineScope = DefaultMapLoaderCoroutineScope, ): MapLoaderAsync<ID, E> { override fun load(id: ID): CompletionStage<E?> = scope.async { suspendTransaction { loadByIdFromDB(id) } }.asCompletableFuture() override fun loadAllKeys(): AsyncIterator<ID> { // loadAllIdsFromDB -> send to channel … return object: AsyncIterator<ID> { // receive from channel override fun hasNext(): CompletionStage<Boolean?> = ensurePending() .thenApply { result -> result.isSuccess } override fun next(): CompletionStage<ID> = ensurePending() .thenApply { result -> pendingReceive = null result.getOrNull() ?: throw NoSuchElementException("No more elements") } } } } R2dbcEntityMapLoader with Exposed R2dbc R2dbcEntityMapLoader.kt R2dbcExposedEntityMapLoader.kt
17. MapWriterAsync<ID, E> R2dbcEntityMapWriter with Exposed R2dbc open class R2dbcEntityMapWriter<ID: Any, E: HasIdentifier<ID>>( private val writeToDb: suspend (map: Map<ID, E>) -> Unit, private val deleteFromDb: suspend (keys: Collection<ID>) -> Unit, private val scope: CoroutineScope = defaultMapWriterCoroutineScope, ): MapWriterAsync<ID, E> { override fun write(map: Map<ID, E>): CompletionStage<Void> = scope.async { suspendTransactionAsync(context = scope.coroutineContext) { writeToDb(map) }.await() null }.asCompletableFuture() override fun delete(ids: Collection<ID>): CompletionStage<Void> = scope.async { suspendTransactionAsync(context = scope.coroutineContext) { deleteFromDb(ids) }.await() null }.asCompletableFuture() } R2dbcEntityMapWriter.kt R2dbcExposedEntityMapWriter.kt
18. R2dbcCachedRepository interface R2dbcCacheRepository<T: HasIdentifier<ID>, ID: Any> { val cacheName: String val entityTable: IdTable<ID> suspend fun ResultRow.toEntity(): T val cache: RMap<ID, T?> suspend fun exists(id: ID): Boolean = cache.containsKeyAsync(id).coAwait() suspend fun findFreshById(id: ID): T? = entityTable.selectAll().where { entityTable.id eq id }.singleOrNull()?.toEntity() suspend fun findFreshAll(vararg ids: ID): List<T> = entityTable.selectAll().where { entityTable.id inList ids.toList() }.map { it.toEntity() }.toList() suspend fun findFreshAll(ids: Collection<ID>): List<T> = entityTable.selectAll().where { entityTable.id inList ids }.map { it.toEntity() }.toList() suspend fun findAll(limit: Int? = null, offset: Long? = null, sortBy: Expression<*> = entityTable.id, sortOrder: SortOrder = SortOrder.ASC, where: SqlExpressionBuilder.() -> Op<Boolean> = { Op.TRUE }, ): List<T> suspend fun get(id: ID): T? = cache.getAsync(id).coAwait() suspend fun getAll(ids: Collection<ID>, batchSize: Int = DefaultBatchSize): List<T> suspend fun put(entity: T): Boolean? = cache.fastPutAsync(entity.id, entity).coAwait() suspend fun putAll(entities: Collection<T>, batchSize: Int = DefaultBatchSize) { cache.putAllAsync(entities.associateBy { it.id }, batchSize).coAwait() } suspend fun invalidate(vararg ids: ID): Long = cache.fastRemoveAsync(*ids).coAwait() suspend fun invalidateAll(): Boolean = cache.clearAsync().coAwait() suspend fun invalidateByPattern(patterns: String, count: Int = DefaultBatchSize): Long { val keys = cache.keySet(patterns, count) return cache.fastRemoveAsync(*keys.toTypedArray()).coAwait() } } R2dbcCacheRepository
19. Create Cache<K,V> for Asynchronous localCachedMap(cacheName, redissonClient) { if (config.isReadOnly) { loaderAsync(r2dbcEntityMapLoader) } else { loaderAsync(r2dbcEntityMapLoader) r2dbcEntityMapWriter.requireNotNull("mapWriter") writerAsync(r2dbcEntityMapWriter) writeMode(config.writeMode) } codec(config.codec) syncStrategy(config.nearCacheSyncStrategy) writeRetryAttempts(config.writeRetryAttempts) writeRetryInterval(config.writeRetryInterval) timeToLive(config.ttl) if (config.nearCacheMaxIdleTime > Duration.ZERO) { maxIdle(config.nearCacheMaxIdleTime) } } mapCache(cacheName, redissonClient) { if (config.isReadOnly) { loaderAsync(r2dbcEntityMapLoader) } else { loaderAsync(r2dbcEntityMapLoader) r2dbcEntityMapWriter.requireNotNull("suspendedMapWriter") writerAsync(r2dbcEntityMapWriter) writeMode(config.writeMode) } codec(config.codec) writeRetryAttempts(config.writeRetryAttempts) writeRetryInterval(config.writeRetryInterval) }.apply { if (config.nearCacheMaxSize > 0) { setMaxSize(config.nearCacheMaxSize, EvictionMode.LRU) } } AbstractR2dbcCacheRepository
20. Using Caches
21. Read Through withR2dbcEntityTable(testDB) { val id = getExistingId() // DBীࢲ ઑഥೠ ч val entityFromDB = repository.findFreshById(id) entityFromDB.shouldNotBeNull() // நदীࢲ ઑഥೠ ч val entityFromCAche = repository.get(id) entityFromCAche.shouldNotBeNull() entityFromCAche shouldBeEqualTo entityFromDB repository.exists(id).shouldBeTrue() } withR2dbcEntityTable(testDB) { val entities = repository.findAll() entities.shouldNotBeEmpty() entities.size shouldBeEqualTo repository.entityTable.selectAll().count().toInt() } R2dbcReadThroughScenario
22. Write Through withR2dbcEntityTable(testDB) { val id = getExistingId() // நदীࢲ ઑഥೠ ч val entity = repository.get(id) entity.shouldNotBeNull() // நदী јनػ ч ੷੢ -> DBীب ੷੢ val updatedEntity = updateEntityEmail(entity) repository.put(updatedEntity) // நदীࢲ ઑഥೠ ч val entityFromCache = repository.get(id) entityFromCache.shouldNotBeNull() assertSameEntityWithoutAudit(entityFromCache, updatedEntity) delay(DEFAULT_DELAY) // DBীࢲ ઑഥೠ ч val entityFromDB = repository.findFreshById(id) entityFromDB.shouldNotBeNull() assertSameEntityWithoutAudit(entityFromDB, entityFromCache) } R2dbcWriteThroughScenario
23. Write Behind withR2dbcEntityTable(testDB) { val entities = createNewEntities(1000) repository.putAll(entities) await .atMost(Duration.ofSeconds(10)) .withPollInterval(Duration.ofMillis(1000)) .coUntil { getAllCountFromDB() >= entities.size.toLong() } // DBীࢲ ઑഥೠ ч val dbCount = getAllCountFromDB() dbCount shouldBeGreaterThan entities.size.toLong() } R2dbcWriteBehindScenario
24. Resources
25. Resources • Bluetape4k-Exposed-R2dbc-Redisson Source • Cache Strategies with Redisson & Exposed R2dbc • Exposed ebook