기존 Spring MVC에서는 @Cacheable 어노테이션을 이용하여 캐싱을 할 수 있지만, Spring Webflux의 경우에는 리턴하는 Mono 객체가 캐싱된다.

=> 리턴하는 Mono 내부의 값을 참조하기 위해서는 직접 구현이 필요!

TODO: cache(timeout: 10min)

의존성 추가

• cache customizing: spring-boot-starter-aop, reactor-extra
  • reactor-extra는 버전 3.6.0 이후로는 CacheMono / CacheFlux 지원을 중단했다 🫥
  • 참고: https://github.com/reactor/reactor-addons/issues/237
• cache timeout config: spring-boot-starter-cache, caffeine

build.gradle.kts

implementation("org.springframework.boot:spring-boot-starter-webflux")
implementation("org.springframework.boot:spring-boot-starter-aop")
implementation("org.springframework.boot:spring-boot-starter-cache")

// For Java 11 or above, use 3.x otherwise use 2.x.
implementation("com.github.ben-manes.caffeine:caffeine:3.1.6")

// 3.4.7 이상 사용할 경우 CacheMono/CacheFlux 사용 불가
implementation("io.projectreactor.addons:reactor-extra:3.3.8.RELEASE")

Cache Config

• 타임아웃 등 캐쉬 설정을 적용한다.

cacheConfig.kt

import com.github.benmanes.caffeine.cache.Caffeine
import org.springframework.cache.CacheManager
import org.springframework.cache.annotation.CachingConfigurerSupport
import org.springframework.cache.annotation.EnableCaching
import org.springframework.cache.caffeine.CaffeineCacheManager
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import java.util.concurrent.TimeUnit

@EnableCaching
@Configuration
class CacheConfig: CachingConfigurerSupport() {
    @Bean
    fun caffeineConfig(): Caffeine<Any, Any> {
        return Caffeine.newBuilder()
        		// timeout 설정
                .expireAfterWrite(10, TimeUnit.MINUTES)
    }

    @Bean
    fun cacheManager(caffeine: Caffeine<Any, Any>): CacheManager {
        val caffeineCacheManager = CaffeineCacheManager()
        caffeineCacheManager.setCaffeine(caffeine)
        return caffeineCacheManager
    }
}

Annotation 만들기

cache/ReactorCacheable.kt

@Target(AnnotationTarget.FUNCTION)
@Retention(AnnotationRetention.RUNTIME)
annotation class ReactorCacheable(
    val name: String = ""
)

Cache Manager

cache/ReactorCacheManager.kt

package com.todolist.cache

import org.springframework.cache.CacheManager
import org.springframework.stereotype.Component
import reactor.cache.CacheFlux
import reactor.cache.CacheMono
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.Signal
import java.util.function.Supplier

@Component
class ReactorCacheManager(
        val cacheManager: CacheManager
) {
    fun findCachedMono(cacheName: String, key: Any, retriever: Supplier<Mono<Any>>, classType: Class<*>?): Mono<Any> {
        val cache = cacheManager.getCache(cacheName)

        assert(cache != null)

        return CacheMono
                .lookup<Any, Any>({ k ->
                    val result = cache!!.get(k, classType)
                    Mono.justOrEmpty(result).map { Signal.next(it) }
                }, key)
                .onCacheMissResume(Mono.defer(retriever))
                .andWriteWith { k, signal ->
                    Mono.fromRunnable {
                        if (!signal.isOnError) {
                            cache!!.put(k, signal.get())
                        }
                    }
                }
    }

    fun findCachedFlux(cacheName: String, key: Any, retriever: Supplier<Flux<Any>>): Flux<Any> {
        val cache = cacheManager.getCache(cacheName)

        assert(cache != null)

        return CacheFlux
                .lookup<Any, Any>({ k ->
                    val result = cache!!.get(k, List::class.java)
                    Mono.justOrEmpty(result).flatMap { list ->
                        Flux.fromIterable(list).materialize().collectList()
                    }
                }, key)
                .onCacheMissResume(Flux.defer(retriever))
                .andWriteWith { k, signalList ->
                    Flux.fromIterable(signalList)
                            .dematerialize<Any>()
                            .collectList()
                            .doOnNext { list ->
                                cache!!.put(k, list)
                            }
                            .then()
                }
    }
}

ReactorCacheAspect

cache/ReactorCacheAspect.kt

package com.todolist.cache

import org.aspectj.lang.ProceedingJoinPoint
import org.aspectj.lang.annotation.Around
import org.aspectj.lang.annotation.Aspect
import org.aspectj.lang.reflect.MethodSignature
import org.springframework.core.ResolvableType
import org.springframework.stereotype.Component
import java.lang.reflect.ParameterizedType
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.*
import java.util.stream.Collectors

@Aspect
@Component
class ReactorCacheAspect(
        private val reactorCacheManager: ReactorCacheManager
) {
    @Around("@annotation(ReactorCacheable)")
    fun around(joinPoint: ProceedingJoinPoint): Any {
        val signature = joinPoint.signature as MethodSignature
        val method = signature.method

        val parametrizedType = method.genericReturnType as ParameterizedType
        val rawType = parametrizedType.rawType

        if (!rawType.equals(Mono::class.java) && !rawType.equals(Flux::class.java)) {
            throw IllegalArgumentException("The return type is not Mono/Flux. Use Mono/Flux for return type. method: " + method.name);
        }

        val reactorCacheable = method.getAnnotation(ReactorCacheable::class.java)
        val cacheName = reactorCacheable.name
        val args = joinPoint.args

        if (rawType.equals(Mono::class.java)) {
            val returnTypeInsideMono = parametrizedType.actualTypeArguments[0]
            val returnClass = ResolvableType.forType(returnTypeInsideMono).resolve()
            val retriever = { joinPoint.proceed(args) as Mono<Any> }

            return reactorCacheManager
                    .findCachedMono(cacheName, generateKey(args), retriever, returnClass)
                    .doOnError { e ->
                        // do something
                    }
        } else {
            val retriever = { joinPoint.proceed(args) as Flux<Any> }

            return reactorCacheManager
                    .findCachedFlux(cacheName, generateKey(args), retriever)
                    .doOnError { e ->
                        // do something
                    }

        }
    }

    private fun generateKey(vararg objects: Any): String {
        return Arrays.stream(objects)
                .map { obj ->
                    obj?.toString() ?: ""
                }
                .collect(Collectors.joining(":"))
    }
}

적용예시

이제 @ReactorCacheable을 붙이면 캐싱됨!!

    @ReactorCacheable
    fun test(): Mono<String> {
        val date = LocalDateTime.now().toString()
        return Mono.just(date)
    }

File Tree

├── build.gradle.kts
└── src
    └── main
        ├── kotlin.com
        │   ├── service
        │   ├── controller
        │   ├── cache
        │   │   ├── ReactorCacheAspect.kt
        │   │   ├── ReactorCacheManager.kt
        │   │   └── ReactorCacheable.kt
        │   └── config
        │       └── CacheConfig.kt
        └── resources
            └── application.yml

참고:
https://devmingsa.tistory.com/81
https://dreamchaser3.tistory.com/17
https://github.com/ben-manes/caffeine