동기·suspend JCache

factory로 시작하기

LettuceCaches는 가장 자주 쓰는 JCache 생성 경로를 모읍니다. 동기 API는 JCache<K, V>, coroutine API는 String key를 쓰는 LettuceSuspendJCache<V>를 반환합니다.

val sessions = LettuceCaches.jcache<String, Session>(
    redisClient,
    cacheName = "sessions-v1",
    ttlSeconds = 1_800,
)

val suspended = LettuceCaches.suspendJCache<Session>(
    redisClient,
    cacheName = "sessions-suspend-v1",
    ttlSeconds = 1_800,
)

두 경로 모두 Redis hash 기반 LettuceJCache를 사용합니다. suspend wrapper는 별도 non-blocking Redis 구현이 아니라 동기 JCache 호출을 Dispatchers.IO에서 실행합니다.

CRUD와 batch

getAll은 key를 100개씩 나누고 hash field를 batch 조회합니다. putAll은 listener가 있을 때만 기존 field를 한 번 읽어 CREATED와 UPDATED를 구분한 뒤 batch write합니다.

sessions.putAll(mapOf("a" to sessionA, "b" to sessionB))
val found = sessions.getAll(setOf("a", "b", "missing"))

check(found.keys == setOf("a", "b"))
check(sessions.putIfAbsent("a", another) == false)

replace(key, old, new)와 remove(key, old)는 읽고 비교한 뒤 별도 Redis 명령을 수행합니다. 여러 client가 같은 key를 동시에 바꾸는 상황에서 원자적 compare-and-set을 보장하지 않습니다. 원자적 조건 변경은 Near Cache의 Lua CAS나 별도 Redis script를 사용합니다.

listener와 EntryProcessor

cache entry listener는 이 LettuceJCache instance가 수행한 CREATED·UPDATED·REMOVED 이벤트를 process 안에서 호출합니다. Redis의 다른 client가 값을 바꿔도 JCache listener가 자동으로 알림을 받는 구조는 아닙니다.

1.11.0의 invoke와 invokeAll은 MutableEntry를 구현합니다. processor가 값을 읽고 setValue 또는 remove를 요청하면 처리 뒤 JCache write로 commit합니다.

val updated = sessions.invoke("a", EntryProcessor<String, Session, Int> { entry, _ ->
    val next = entry.value!!.copy(refreshCount = entry.value!!.refreshCount + 1)
    entry.setValue(next)
    next.refreshCount
})

이 과정도 Redis transaction이나 Lua script 하나로 묶이지 않습니다. 같은 key에 대한 외부 동시 변경과 충돌할 수 있으므로 분산 원자 연산으로 해석하지 않습니다.

suspend API의 실행 경계

LettuceSuspendJCache는 get, put, remove, replace, close를 Dispatchers.IO에서 호출합니다. getAll(keys)와 entries()는 Flow를 반환하지만 내부 source는 동기 JCache 조회입니다.

coroutineScope {
    suspended.put("a", sessionA)
    suspended.getAll(setOf("a", "b")).collect { entry ->
        audit(entry.key, entry.value)
    }
}

putAllFlow는 입력 Flow를 하나씩 수집해 put합니다. 대량 입력을 Redis 한 번의 batch로 보내는 API가 아니므로 throughput이 중요하면 map을 모아 putAll을 사용합니다.

manager와 종료

LettuceSuspendCacheManager는 이름별 wrapper를 재사용하고 기본 TTL·codec을 적용합니다. closeCache는 registry에서 wrapper를 제거하지만 Redis hash는 남깁니다. 새 wrapper를 열면 같은 데이터를 다시 읽을 수 있습니다.

val manager = LettuceSuspendCacheManager(redisClient, defaultTtlSeconds = 600)
val cache = manager.getOrCreate<User>("users-v1")

cache.put("42", user)
manager.closeCache(cache)
check(manager.getOrCreate<User>("users-v1").get("42") == user)

데이터까지 삭제하려면 destroyCache 또는 clear를 명시적으로 호출합니다. manager의 suspend close()는 등록 cache를 모두 닫고, cache close 중 CancellationException이 발생해도 나머지 정리를 시도한 뒤 취소를 다시 전파합니다.