#17: Optimising your code with WebFlux, p. 1

🕐 Reading time: 3 minutes

In the AML project we're currently assigned to, good old Mario had to improve the performance of some services, avoiding the fetching of data that didn't change frequently. While exploring the WebFlux documentation, he found the .cache() and .share() methods, which lend themselves nicely to a new article!

📌 Cold Flux and why you need `.cache()`

In Reactor, Fluxes are cold: every time a subscriber subscribes, the flow starts over from scratch, re-running the logic at the source. This is the standard behaviour, but not always the desirable one. Imagine, for example, several subscribers that, within a few minutes of each other, query the same expensive source (e.g. a configuration table that updates every 10 minutes). In these cases, re-running everything each time is inefficient. The .cache() method lets you store the data emitted by the flow on the first subscription, and reuse it for the following ones, thus avoiding recomputation.

📑 The experiment – with and without cache

class SampleTest017 {

    @Test
    void test_buildFlux_noCache() {
        final AtomicInteger subscriptionCount = new AtomicInteger();
        final Flux<String> sharedFlux = buildFlux(subscriptionCount);
        final List<String> result1 = invokeAndCollect(sharedFlux);
        final List<String> result2 = invokeAndCollect(sharedFlux);
        checkResults(result1, result2);
        //expect <2> because the source is cold and not cached
        assertEquals(2, subscriptionCount.get());
    }

    @Test
    void test_buildFlux_withCache() {
        final AtomicInteger subscriptionCount = new AtomicInteger();
        final Flux<String> sharedFlux = buildFlux(subscriptionCount)
                .cache(); //cache the emitted items for all subsequent subscribers
        final List<String> result1 = invokeAndCollect(sharedFlux); //1st subscription triggers the source
        final List<String> result2 = invokeAndCollect(sharedFlux); //2nd subscription uses the cached data
        checkResults(result1, result2);
        //expect <1> because the result is cached after the first subscription
        assertEquals(1, subscriptionCount.get());
    }

    //build a Flux that emits "A", "B", "C" and increments the count each time the Flux is subscribed to
    private Flux<String> buildFlux(final AtomicInteger count) {
        return Flux.just("A", "B", "C").doOnSubscribe(subscription -> count.incrementAndGet());
    }

    private List<String> invokeAndCollect(final Flux<String> sharedFlux) {
        return sharedFlux.collectList().block();
    }

    @SafeVarargs
    private void checkResults(final List<String>... results) {
        Arrays.stream(results).forEach(result -> assertEquals(List.of("A", "B", "C"), result));
    }
}

📑 Analysis of the 1st test – `test_buildFlux_noCache()`

The default behaviour is clearly visible:

the Flux emits the values "A", "B", "C" and increments a counter every time a new subscription starts
two calls to invokeAndCollect(...) amount to two distinct executions

📑 Analysis of the 2nd test – `test_buildFlux_withCache()`

Here we use .cache():

the first subscription activates the flow and stores its values
the following subscriptions get the already-emitted data, without re-running it (and subscriptionCount isn't incremented)

📌 When is it worth using `cache`?

when the data doesn't change often
if the subscribers arrive close together in time
if we want to avoid latency, needless load or wasted resources

⚠️ Limits and things to watch out for

the cache is in memory and local to the Flux: if you recreate the Flux, you lose the cache
multiple threads accessing the same Flux object with .cache() share the data, but it's not a globally shared cache
by default, the cache doesn't expire: the data stays until the Flux is recreated or the garbage collector cleans it up. You can, however, specify a timeout with .cache(Duration).
cache shouldn't be confused with Spring's @Cacheable annotation: cache is a reactive method, local and temporary; whereas @Cacheable is centralised and designed for distributed or persistent caches, based on keys
cache is better suited to finite flows: if used with infinite flows, it risks taking up more and more memory, unless there are explicit limitations like .take(n)

This article is getting way too long… and after all, the rubric is called "Pills"! I'll stop here and leave the .share() method for part 2 – stay tuned!

See you at the next pill! ☕

Java Pills