Data caching is a widespread technique in the programming. It allows to quickly retrieve data without making long-running operations. But there is a problem with caching of data retrieved as result of some long-running operation. If a cache value is missed, it will be requested. If it is requested by a long-running HTTP request or SQL command, the next request for the cache value can leads to multiple HTTP requests / SQL commands again and again. I was looking for a cache implementation which solves this issue in projects using Project Reactor. Project Reactor is built on top of the Reactive Streams Specification — a standard for building reactive applications. You probably know Mono and Flux objects from Spring WebFlux. Project Reactor is the reactive library of choice for Spring WebFlux.
In this article, I will suggest a reactive cache implementation inspired by CacheMono from Reactor’s addons project. We will assume, that the result of a long-running HTTP request or SQL command is represented as aMono object. A Mono object is “materialized” and cached in form of Reactor’s Signal object which represents a Mono. Signals are “dematerialized” to Mono’s if a cache value is requested by the lookup method. Multiple lookups with the same key will retieve the same Mono object, so that a long-running operation is only triggered once!
Let’s create a class CacheMono with three factory methods.
@Slf4j
public class CacheMono<KEY, IVALUE, OVALUE> {
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private final Map<KEY, CacheMonoValue<OVALUE>> cache = new HashMap<>();
/**
* External value supplier which should be provided if "valuePublisher" with "keyExtractor"
* are not set
*/
private final Function<KEY, Mono<OVALUE>> valueSupplier;
/**
* External source publisher stream which should be provided if "valueSupplier" is not set
*/
private final Flux<IVALUE> valuePublisher;
/**
* Key extractor for emitted items provided by "valuePublisher"
*/
private final Function<IVALUE, KEY> keyExtractor;
/**
* Value extractor for emitted items provided by "valuePublisher"
*/
private final Function<IVALUE, OVALUE> valueExtractor;
private CacheMono(Function<KEY, Mono<OVALUE>> valueSupplier, Flux<IVALUE> valuePublisher,
Function<IVALUE, KEY> keyExtractor, Function<IVALUE, OVALUE> valueExtractor) {
this.valueSupplier = valueSupplier;
this.valuePublisher = valuePublisher;
this.keyExtractor = keyExtractor;
this.valueExtractor = valueExtractor;
}
/**
* Factory method to create a CacheMono instance from an external value supplier. The value
* supplier is called by this CacheMono instance for retrieving values when they are missing
* in cache ("pull" principle to retrieve not yet cached values).
*/
public static <KEY, VALUE> CacheMono<KEY, VALUE, VALUE> fromSupplier(
@NonNull Function<KEY, Mono<VALUE>> valueSupplier) {
Objects.requireNonNull(valueSupplier);
return new CacheMono<>(valueSupplier, null, null, null);
}
/**
* Factory method to create a CacheMono instance from an external value publisher.
* Published values will fill this cache (reactive "push" way).
*/
public static <KEY, VALUE> CacheMono<KEY, VALUE, VALUE> fromPublisher(
@NonNull Flux<VALUE> valuePublisher, @NonNull Function<VALUE, KEY> keyExtractor) {
Objects.requireNonNull(valuePublisher);
Objects.requireNonNull(keyExtractor);
return createCacheMono(valuePublisher, keyExtractor, Function.identity());
}
/**
* Factory method to create a CacheMono instance from an external value publisher.
* Published values will fill this cache (reactive "push" way).
*/
public static <KEY, IVALUE, OVALUE> CacheMono<KEY, IVALUE, OVALUE> fromPublisher(
@NonNull Flux<IVALUE> valuePublisher,
@NonNull Function<IVALUE, KEY> keyExtractor,
@NonNull Function<IVALUE, OVALUE> valueExtractor) {
Objects.requireNonNull(valuePublisher);
Objects.requireNonNull(keyExtractor);
return createCacheMono(valuePublisher, keyExtractor, valueExtractor);
}
private static <KEY, IVALUE, OVALUE> CacheMono<KEY, IVALUE, OVALUE> createCacheMono(
@NonNull Flux<IVALUE> valuePublisher,
@NonNull Function<IVALUE, KEY> keyExtractor,
@NonNull Function<IVALUE, OVALUE> valueExtractor) {
var cacheMono = new CacheMono<>(null, valuePublisher, keyExtractor, valueExtractor);
valuePublisher.doOnEach(signal -> {
if (signal.hasValue()) {
final var inputValue = signal.get();
final var outputSignal = Signal.next(valueExtractor.apply(inputValue));
cacheMono.cache.put(keyExtractor.apply(inputValue),
new CacheMonoValue<>(outputSignal));
} else if (signal.isOnError()) {
if (signal.getThrowable() == null) {
log.error("Error from value publisher");
} else {
log.error("Error from value publisher, message = {}",
signal.getThrowable().getMessage());
}
}
}).subscribe();
return cacheMono;
}
...
}
Not yet cached values will be retrieved either by valueSupplier or valuePublisher. The first one uses the “pull” principle and the second one uses the “push” principle to retrieve not yet cached values. That means, either valueSupplier or valuePublisher along with keyExtractor and valueExtractor should be set.
Keep in mind: if you create more than one CacheMono from the same value publisher, you should pass in a Flux stream which caches the history and emits cached items from the beginning to future subscribers. This is necessary because this CacheMono implementation subscribes to the passed in Flux stream in order to fill cache automatically once the source Flux stream publishes values (reactive “push” way vs. “pull” provided by another factory method). The simplest way to create a such Flux stream from existing one would be invoking of cache() method on any Flux stream.
As you could see, we cache instances of CacheMonoValue. This is just a wrapper around Mono or Signal. We can implement this class as an inner class.
private static class CacheMonoValue<VALUE> {
private Mono<VALUE> mono;
private Signal<VALUE> signal;
CacheMonoValue(Mono<VALUE> mono) {
this.mono = mono;
}
CacheMonoValue(Signal<VALUE> signal) {
this.signal = signal;
}
Mono<VALUE> toMono() {
if (mono != null) {
return mono;
}
return Mono.justOrEmpty(signal).dematerialize();
}
Optional<VALUE> getValue() {
if (signal == null) {
return Optional.empty();
}
return Optional.ofNullable(signal.get());
}
}
We will see in few words, that a Mono value from a long-running operation is cached immediately. The same Mono instance is retrieved for all subsequent lookups with the same key. Once the result of Mono is available, the real value is cached as Signal under the same key. Well, step by step. Look at the lookup method first. It uses a well-known pattern: if value is missed in the cache, the logic within the switchIfEmpty operator gets executed.
/**
* Finds a value by key in an in-memory cache or load it from a remote source.
* The loaded value will be cached.
*/
public Mono<OVALUE> lookup(KEY key) {
return Mono.defer(() -> getValueAsMono(key)
.switchIfEmpty(Mono.defer(() -> onCacheMissResume(key)))
);
}
private Mono<OVALUE> getValueAsMono(KEY key) {
final Lock readLock = lock.readLock();
readLock.lock();
try {
return Mono.justOrEmpty(cache.get(key)).flatMap(CacheMonoValue::toMono);
} finally {
readLock.unlock();
}
}
private Mono<OVALUE> onCacheMissResume(KEY key) {
final Lock writeLock = lock.writeLock();
writeLock.lock();
try {
// check if value was already cached by another thread
final var cachedValue = cache.get(key);
if (cachedValue == null) {
final Mono<OVALUE> monoValue;
if (valuePublisher != null) {
// get value from external value publisher
monoValue = valuePublisher
.filter(value -> Objects.equals(keyExtractor.apply(value), key))
.map(valueExtractor)
.next();
} else if (valueSupplier != null) {
// get value from external supplier
monoValue = valueSupplier.apply(key);
} else {
throw new IllegalStateException("Value can be not determined," +
"neither valuePublisher nor valueSupplier were set");
}
// cache Mono as value immediately
cache.put(key, new CacheMonoValue<>(monoValue));
// cache success and error values encapsulated in signal when it is available
return monoValue.doOnEach(signal -> {
if (signal.isOnNext()) {
cache.put(key, new CacheMonoValue<>(
Signal.next(Objects.requireNonNull(signal.get())))
);
} else if (signal.isOnError()) {
final Signal<OVALUE> errorSignal;
if (signal.getThrowable() == null) {
errorSignal = Signal.error(
new Throwable("Getting value from external provider failed"));
} else {
errorSignal = Signal.error(signal.getThrowable());
}
cache.put(key, new CacheMonoValue<>(errorSignal));
}
});
}
return Mono.justOrEmpty(cachedValue).flatMap(CacheMonoValue::toMono);
} finally {
writeLock.unlock();
}
}
In the onCacheMissResume, a missed value will be retieved by the mentioned above valueSupplier or valuePublisher. As I said, the value is cached immediately as a Mono object and is returned for all subsequent lookups. As soon as the value from the long-running operation is available, the logic within monoValue.doOnEach(...) is executed. The value is encapsulated in Signal and can be returned by invokingsignal.get().
Let’s implement some convenient methods as well. Especially methods which return already existing (cached) values from the cache.
/**
* Gets cached values as Java Stream. Returned stream is not sorted.
*/
public Stream<OVALUE> getValues() {
final Lock readLock = lock.readLock();
readLock.lock();
try {
return cache.values().stream().flatMap(cachedValue -> cachedValue.getValue().stream());
} finally {
readLock.unlock();
}
}
/**
* Gets cached value as Java Optional.
*/
public Optional<OVALUE> getValue(KEY key) {
final Lock readLock = lock.readLock();
readLock.lock();
try {
return Optional.ofNullable(cache.get(key)).flatMap(CacheMonoValue::getValue);
} finally {
readLock.unlock();
}
}
/**
* Removes the mapping for a key from this map if it is present.
*/
public void remove(KEY key) {
final Lock writeLock = lock.writeLock();
writeLock.lock();
try {
cache.remove(key);
} finally {
writeLock.unlock();
}
}
The usage of CacheMono class is simple. Just two code snippets from my current project. The first one creates a CacheMono instance by calling CacheMono.fromSupplier.
@Service
@Slf4j
@RequiredArgsConstructor
public class TopologyRepository {
private final CacheMono<TopologyRef, TopologyDto, TopologyDto> cache;
private final TopologyLoader topologyLoader;
private final TopologyCreator topologyCreator;
@Autowired
public UnoTopologyRepository(TopologyLoader topologyLoader,
TopologyCreator topologyCreator) {
this.topologyLoader = topologyLoader;
this.topologyCreator = topologyCreator;
cache = CacheMono.fromSupplier(this::retrieveTopology);
}
/**
* Finds a topology from this repository by reference.
*/
public Mono<TopologyDto> findUnoTopology(TopologyRef topologyRef) {
return cache.lookup(topologyRef)
.doOnNext(topology ->
log.info("Topology was found by lookup with key {}", topologyRef))
.onErrorResume(err -> {
log.error("Error on lookup Topology by key {}, message: {}",
topologyRef, err.getMessage());
return Mono.empty();
});
}
private Mono<TopologyDto> retrieveTopology(TopologyRef topologyRef) {
CompletableFuture<UnoTopologyDto> future = CompletableFuture.supplyAsync(() -> {
final var loaderContext = topologyLoader.retrieveTopology(topologyRef);
return topologyCreator.createTopology(loaderContext);
});
return Mono.fromFuture(future);
}
}
The second one creates a CacheMono instance by calling CacheMono.fromPublisher.
@Service
@Slf4j
@RequiredArgsConstructor
public class SspDefinitionenStore implements SspDefinitionConsumer {
private CacheMono>VersionedId, SspDefinition, SspDefinition> sspDefinitionCache;
private FluxSink>SspDefinition> sspDefinitionSink;
@PostConstruct
public void initialize() {
sspDefinitionCache = CacheMono.fromPublisher(
Flux.create(sink -> sspDefinitionSink = sink),
SspDefinition::getId);
}
@Override
public void accept(SspDefinition sspDefinition) {
sspDefinitionSink.next(sspDefinition);
}
public Mono>SspDefinition> lookupSspDefinition(VersionedId sspId) {
return sspDefinitionCache.lookup(sspId)
.doOnNext(sspTopology -> log.info(
"SspDefinition was found by lookup with key {}", sspId))
.onErrorResume(err -> {
log.error("Error on lookup SspDefinition by key {}, message: {}",
sspId, err.getMessage());
return Mono.empty();
});
}
public Optional>SspDefinition> findSspDefinition(VersionedId sspId) {
return sspDefinitionCache.getValue(sspId);
}
public Flux>SspDefinition> findSspDefinitions() {
return Flux.fromStream(sspDefinitionCache.getValues().filter(Objects::nonNull));
}
...
}That’s all. Have fun!
Published on Java Code Geeks with permission by Oleg Varaksin, partner at our JCG program. See the original article here: Clever cache for Reactor’s Mono objects Opinions expressed by Java Code Geeks contributors are their own. |
