cold, hot 이라는 표현을 사용한다.

• cold는 subscribe 실행시마다 파이프라인이 전부 다 새로 실행되는 파이프라인의 상태를 말한다.
• hot은 결과가 캐시되어 공유될 수 있는 상태를 말한다.

// cold: subscribe마다 새로 실행
val cold = Mono.fromCallable {
    println("DB 조회!")
    "result"
}

cold.subscribe()  // "DB 조회!" 출력
cold.subscribe()  // "DB 조회!" 출력
cold.subscribe()  // "DB 조회!" 출력

// hot: 결과 공유
val hot = cold.cache()

hot.subscribe()  // "DB 조회!" 출력
hot.subscribe()  // 출력 없음 (캐시)
hot.subscribe()  // 출력 없음 (캐시)

주의점은 파이프라인에서 발생한 에러도 캐시된다는 것임. 그래서 첫 실행에서 에러 발생시 이후 subscribe에서도 계속 같은 에러가 반환된다.

이를 방지하기 위해 다음과 같은 설정을 추가할 수 있다.

// 에러 캐싱 방지
mono.cache(
    { Duration.ofMinutes(5) },   // 값은 5분 캐싱
    { Duration.ZERO },            // 에러는 캐싱 안 함
    { Duration.ZERO }             // empty도 캐싱 안 함
)

캐시 TTL도 설정 가능하다.

// 무기한 캐싱
mono.cache()

// 단순 TTL
mono.cache(Duration.ofMinutes(5))

// 값/에러/empty 각각 다른 TTL
mono.cache(
    { value -> Duration.ofMinutes(10) },  // 값은 10분
    { error -> Duration.ofSeconds(30) },  // 에러는 30초
    { Duration.ofMinutes(1) }             // empty는 1분
)

flux는 내보낸 값들의 히스토리를 캐싱한다. 히스토리의 크기 제한도 가능함.

// Flux는 방출한 값들의 히스토리를 캐싱
val cachedFlux = Flux.just(1, 2, 3).cache()

cachedFlux.subscribe { println("첫 번째: $it") }  // 1, 2, 3
cachedFlux.subscribe { println("두 번째: $it") }  // 1, 2, 3 (캐시)

// 히스토리 크기 제한
val limitedCache = flux.cache(2)  // 최근 2개만 캐싱

더 정교하게 캐시하고 싶다면 reactor-extra의 CacheMono를 사용한다.

// Redis나 Caffeine 같은 외부 캐시와 연동
val cachedUser = CacheMono
    .lookup(
        { key -> redisCache.get(key).map { Signal.next(it) } },  // 캐시 조회
        userId.toString()
    )
    .onCacheMissResume { userService.find(userId) }  // 캐시 미스 시
    .andWriteWith { key, signal ->
        redisCache.put(key, signal.get())  // 캐시 저장
    }