Skip to main content


Monitoring and measuring reactive application with Dropwizard Metrics

In the previous article we created a simple indexing code that hammers ElasticSearch with thousands of concurrent requests. The only way to monitor the performance of our system was an old-school logging statement:

.window(Duration.ofSeconds(1)) .flatMap(Flux::count) .subscribe(winSize -> log.debug("Got {} responses in last second", winSize)); It's fine, but on a production system, we'd rather have some centralized monitoring and charting solution for gathering various metrics. This becomes especially important once you have hundreds of different applications in thousands of instances. Having a single graphical dashboard, aggregating all important information, becomes crucial. We need two components in order to collect some metrics:

publishing metricscollecting and visualizing them Publishing metrics using Dropwizard Metrics In Spring Boot 2 Dropwizard Metrics were replaced by Micrometer. This article uses the former, the next one will show the latter solution in pr…

Spring, Reactor and ElasticSearch: bechmarking with fake test data

In the previous article we created a simple adapter from ElasticSearch's API to Reactor's Mono, that looks like this:

import reactor.core.publisher.Mono; private Mono<IndexResponse> indexDoc(Doc doc) { //... } Now we would like to run this method at controlled concurrency level, millions of times. Basically, we want to see how our indexing code behaves under load, benchmark it.

Fake data with jFairy First, we need some good looking test data. For that purpose, we'll use a handy jFairy library. The document we'll index is a simple POJO:

@Value class Doc { private final String username; private final String json; } The generation logic is wrapped inside a Java class:

import io.codearte.jfairy.Fairy; import io.codearte.jfairy.producer.person.Address; import io.codearte.jfairy.producer.person.Person; import org.apache.commons.lang3.RandomUtils; @Component class PersonGenerator { private final ObjectMapper objectMapper; private final Fairy fairy;…

Spring, Reactor and ElasticSearch: from callbacks to reactive streams

Spring 5 (and Boot 2, when it arrives in a couple of weeks) is a revolution. Not the "annotations over XML" or "Java classes over annotations" type of revolution. It's truly a revolutionary framework that enables writing a brand new class of applications. Over the recent years, I became a little bit intimidated by this framework. "Spring Cloud being framework that simplifies the usage of Spring Boot, being a framework that simplifies the usage of Spring, being a framework, that simplifies enterprise development." (also known as "start... dot spring... dot I... O") lists 120 different modules (!) that you can add to your service. Spring these days became an enormous umbrella project and I can imagine why some people (still!) prefer Java EE (or whatever it's called these days).

But Spring 5 brings the reactive revolution. It's no longer only a wrapper around blocking servlet API and various web frameworks. Spring 5, on…

Idiomatic concurrency: flatMap() vs. parallel() - RxJava FAQ

Simple, effective and safe concurrency was one of the design principles of RxJava. Yet, ironically, it's probably one of the most misunderstood aspects of this library. Let's take a simple example: imagine we have a bunch of UUIDs and for each one of them we must perform a set of tasks. The first problem is to perform I/O intensive operation per each UUID, for example loading an object from a database:

Flowable<UUID> ids = Flowable .fromCallable(UUID::randomUUID) .repeat() .take(100); ids.subscribe(id -> slowLoadBy(id)); First I'm generating 100 random UUIDs just for the sake of testing. Then for each UUID I'd like to load a record using the following method:

Person slowLoadBy(UUID id) { //... } The implementation of slowLoadBy() is irrelevant, just keep in mind it's slow and blocking. Using subscribe() to invoke slowLoadBy() has many disadvantages:

subscribe() is single-threaded by design and there is no way around it. Each UUID…

Detecting and testing stalled streams - RxJava FAQ

Imagine you have a stream that publishes events with unpredictable frequency. Sometimes you can expect dozens of messages per second, but occasionally no events can be seen for several seconds. This can be an issue if your stream is transmitted over web socket, SSE or any other network protocol. Silent period taking too long (stall) can be interpreted as network issue. Therefore we often send artificial events (pings) once in a while just to make sure:

clients are still alivelet clients know we are still alive A more concrete example, imagine we have a Flowable<String> stream that produces some events. When there is no event for more than one second, we should send a placeholder "PING" message. When the silence is even longer, there should be a "PING" message every second. How can we implement such a requirement in RxJava? The most obvious, but incorrect solution is to merge original stream with pings:

Flowable<String> events = //... Flowable<String&g…

Fixed-rate vs. fixed-delay - RxJava FAQ

If you are using plain Java, since version 5 we have a handy scheduler class that allows running tasks at fixed rate or with fixed delay:

import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10); Basically it supports two types of operations:

scheduler.scheduleAtFixedRate(() -> doStuff(), 2, 1, SECONDS); scheduler.scheduleWithFixedDelay(() -> doStuff(), 2, 1, SECONDS);scheduleAtFixedRate() will make sure doStuff() is invoked precisely every second with an initial delay of two seconds. Of course garbage collection, context-switching, etc. still can affect the precision. scheduleWithFixedDelay() is seemingly similar, however it takes doStuff() processing time into account. For example, if doStuff() runs for 200ms, fixed rate will wait only 800ms until next retry. scheduleWithFixedDelay() on the other hand, always waits for the same amount of time (1 second in our …

Streaming large JSON file with Jackson - RxJava FAQ

In the previous article, we learned how to parse excessively large XML files and turn them into RxJava streams. This time let's look at a large JSON file. We will base our examples on tiny colors.json containing almost 150 records of such format:

{ "aliceblue": [240, 248, 255, 1], "antiquewhite": [250, 235, 215, 1], "aqua": [0, 255, 255, 1], "aquamarine": [127, 255, 212, 1], "azure": [240, 255, 255, 1], //... Little known fact: azure is also a colour and python is a snake. But back to RxJava. This file is tiny but we'll use it to learn some principles. If you follow them you'll be capable of loading and continually processing arbitrarily large, even infinitely long JSON files. First of all the standard "Jackson" way is similar to JAXB: loading the whole file into memory and mapping it to Java beans. However, if your file is in megabyte or gigabytes (because somehow you found JSON to be the best format …

Loading files with backpressure - RxJava FAQ

Processing file as a stream turns out to be tremendously effective and convenient. Many people seem to forget that since Java 8 (3+ years!) we can very easily turn any file into a stream of lines:

String filePath = "foobar.txt"; try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) { reader.lines() .filter(line -> !line.startsWith("#")) .map(String::toLowerCase) .flatMap(line -> Stream.of(line.split(" "))) .forEach(System.out::println); }reader.lines() returns a Stream<String> which you can further transform. In this example, we discard lines starting with "#" and explode each line by splitting it into words. This way we achieve stream of words as opposed to stream of lines. Working with text files is almost as simple as working with normal Java collections. In RxJava we already learned about generate() operator. It can be used here as well to create robust stre…

Generating backpressure-aware streams with Flowable.generate() - RxJava FAQ

RxJava is missing a factory to create an infinite stream of natural numbers. Such a stream is useful e.g. when you want to assign unique sequence numbers to possibly infinite stream of events by zipping both of them:

Flowable<Long> naturalNumbers = //??? Flowable<Event> someInfiniteEventStream = //... Flowable<Pair<Long, Event>> sequenced = naturalNumbers, someInfiniteEventStream, Pair::of ); Implementing naturalNumbers is surprisingly complex. In RxJava 1.x you could briefly get away with Observable that does not respect backpressure:

import rx.Observable; //RxJava 1.x Observable<Long> naturalNumbers = Observable.create(subscriber -> { long state = 0; //poor solution :-( while (!subscriber.isUnsubscribed()) { subscriber.onNext(state++); } }); What does it mean that such stream is not backpressure-aware? Well, basically the stream produces events (ever-incrementing state variable) as fast …

1.x to 2.x migration: Observable vs. Observable: RxJava FAQ

The title is not a mistake. rx.Observable from RxJava 1.x is a completely different beast than io.reactivex.Observable from 2.x. Blindly upgrading rx dependency and renaming all imports in your project will compile (with minor changes) but does not guarantee the same behavior. In the very early days of the project Observable in 1.x had no notion of backpressure but later on backpressure was included. What does it actually mean? Let's imagine we have a stream that produces one event every 1 millisecond but it takes 1 second to process one such item. You see it can't possibly work this way in the long run:

import rx.Observable; //RxJava 1.x import rx.schedulers.Schedulers; Observable .interval(1, MILLISECONDS) .observeOn(Schedulers.computation()) .subscribe( x -> sleep(Duration.ofSeconds(1)));MissingBackpressureException creeps in within few hundred milliseconds. But what does this exception mean? Well, basically it's a safety net…