Integration Support with Spring Webflux ?

[ShanmugamC]

Hi , Can we integrate this caffeine library with Spring Webflux (Reactive) ?

[ben-manes]

Reactor provided this as an add-on, see CacheMono. The JavaDoc shows an example using Caffeine.

This design tries to generalize to a Map<K, V> style cache, which leads to cache stampedes. That has become problematic and they may phase it out (reactor/reactor-addons#237). The proposed replacement follows the strategy that we recommend too,

When using a caching solution... that has a future-based async API, we can recommend to just wrap the futures using Mono#toFuture() and Mono.fromFuture(f).

It might be nice if someone wrote an adapter between our async interfaces using CompletableFuture to a set using Mono for convenience of avoiding this wrap/unwrap logic in every use-case. That might mirror Scaffeine, which offers that for Scala's Future type to make the library feel idiomatic to their ecosystem.

[jaesuk-kim0808]

Hi. I am building a high-performance API service using spring webflux and caffeine.
I was also thinking about the same problem and I am using it like this:

/**
I created CustomCacheManager and annotation to use instead of CacheManager and @Cacheable functions of spring-cache.
AsyncLoadingCache is created at the first call using spring-aop, and after that, take it out of CustomCacheManager and use it.
AsyncLoadingCache is created using Mono.defer(#method that require caching#).toFuture().
*/


   @Pointcut("@annotation(AsyncCacheable)")
    public void pointcut() {
    }

    @Around("pointcut()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();

        AsyncCacheable asyncCacheable = method.getAnnotation(AsyncCacheable.class);
        String cacheName = asyncCacheable.name();
        Object[] args = joinPoint.getArgs();

        AsyncLoadingCache asyncLoadingCache = asyncCacheManager.get(cacheName);
        if (Objects.nonNull(asyncLoadingCache)) {
            return Mono.fromFuture(asyncLoadingCache.get(generateKey(args)));
        }

        CacheType cacheType = CacheType.nameOf(cacheName);
        if (Objects.nonNull(cacheType)) {
            ThrowingSupplier retriever = () -> joinPoint.proceed(args);
            asyncLoadingCache = Caffeine.newBuilder()
                .maximumSize(cacheType.getMaximumSize())
                .expireAfterWrite(cacheType.getExpiredAfterWrite(), TimeUnit.SECONDS)
                .buildAsync((key, exec) -> Mono.defer(retriever).toFuture());
            asyncCacheManager.add(cacheName, asyncLoadingCache);
            log.info("Build AsyncLoadingCache : {}", cacheName);
            return Mono.fromFuture(asyncLoadingCache.get(generateKey(args)));
        }

        return joinPoint.proceed(args);
    }

    @AsyncCacheable(name = "GameGeneralDataProvider.getBySvcId")
    @Override
    public Mono<GameGeneralInfoDomain> getBySvcId(String svcId) {}

Finally, I checked if there is a blocking section using reactor/BlockHound (https://github.com/reactor/BlockHound) and it was fine.

@ben-manes Although I am careful about the question because I implemented the code using spring f/w, is there any section that could be a problem??

[jaesuk-kim0808]

You were right. Originally, it was implemented to create AsycLoadingCache at the time of bean creation using @PostConstruct, put it in the cache manager, and use it. There was no problem this time, but I used spring-aop to reduce unnecessary code writing, and as you said, it is capturing in the first call.
I will think more about this point.

[ben-manes]

The simplest is just to not support refreshAfterWrite as you are not using it in your sample. Then only create an AsyncCache and use it like a computeIfAbsent call,

return cache.get(generateKey(args), (key, exec) -> Mono.defer(joinPoint.proceed(args)).toFuture());

If you do need refreshAfterWrite then then the args need to be decomposed from the key. Since the joinPoint takes those in, I assume it is stateless and you can reuse it. Then the build method be something like,

.build((key, exec) -> Mono.defer(joinPoint.proceed(unpack(key))).toFuture());

[jaesuk-kim0808]

As in the code below, I changed from AsyncLoadingCache to AsyncCacheable and tested it, and it was confirmed that there is no problem.

Mono<?> retVal = Mono.defer(() -> {
	try {
		return (Mono)joinPoint.proceed();
	} catch (Throwable throwable) {
		return Mono.empty();
	}
});
CompletableFuture completableFuture = asyncCache.get(generateKey(args), (key, exec) -> (retVal).toFuture());
return Objects.nonNull(completableFuture) ?
	Mono.fromFuture(completableFuture) : retVal;

The refreshAfterWrite function is not expected to be used in the future, but I'll consider your second idea just in case.
I have one question, is there any performance difference between getIfPresent and get(like a compiteIfAbsent) of AsyncCache?? Most of the time I'll get it with getIfPresent, so is it better to put only on failure?

[ben-manes]

The main reason to prefer computeIfAbsent style methods is to avoid cache stampedes (racy get-load-put that do the load multiple times). While ConcurrentHashMap's computeIfAbsent is more pessimistic, Caffeine's implementation does a getIfPresent call first as a fast path before falling back to a compute-style method if absent. The performance of a populated AsyncCache.get vs AsyncCache.getIfPresent should be nearly identical. Both styles are offered since you'll know best what is most appropriate, but we do advocate that the default practice should be to prefer loading through the cache to avoid those simple, easily overlooked mistakes like a cache stampede.

[jaesuk-kim0808]

I think it's a very good mechanism. I was able to solve the problem with your active support. I'll run the tests further and share if there are any issues. Thank you sincerely.

[shaikezr]

I had been struggling with this for a while and didn't find a solution that worked out of the box for me. After some playing around, I got this solution to work. I took the solutions I saw mentioned in various discussion, mainly this one, and created a fully fleshed out solution including imports and the dependencies you need to add to the project, called @AsyncCacheable.

Example usage is in the ReadMe. It's @jaesuk-kim0808 's solution, plus a custom annotation and the imports + dependencies.

Hopefully somebody finds this useful/it saves you the trouble of trying to implement this 5 times and hitting errors: https://github.com/shaikezr/async-cacheable

Also, in the above solution, the cache is only unique by its name. I changed this to be a hash of the params so each cache with different params will be separate and added params for max size and expireAfterWrite.

Additionally, I made the key of the cached items include a hash of the method so that methods with the same params get different cache keys.

[sachinsinghsindhu]

@shaikezr will this solution cache any errors as well and return error in future calls from cache till its ttl?
https://github.com/shaikezr/async-cacheable/blob/main/AsyncCacheAspect.java#L73
here we are returning a mono in case of error.

[shaikezr]

Yes, you’re right. I haven’t tested that but I imagine it would. We could throw something here instead of returning mono.error

[ben-manes]

fwiw, Caffeine has a whenComplete handler to remove an unsuccessful future. Since it is asynchronous the failed future will reside in the cache for a short duration, and since handlers are attached in a stack ours might be executed last. An unsuccessful future is if it resolves to a null value or was exceptional, as caching either would best be done by a wrapper (e.g. optional). This way it has the closest behavior to a synchronous map / cache as a conceptual model (null values disallowed, exceptional compute does not create a mapping). I think that might mean your mono is discarded proactively, but I am not familiar with their apis to be certain.