Asynchronous and reactive methodologies allow you to utilize better system resources, instead of wasting threads waiting for network or disk I/O. Threads can be fully utilized to perform other work instead.

A broad range of technologies exists to facilitate this style of programming, ranging from the very limited and less usable java.util.concurrent.Future to complete libraries and runtimes like Akka. One library, RxJava, has a very rich set of operators to compose asynchronous workflows, it has no further dependencies to other frameworks and supports the very mature Rx model. For more information about Reactive Extensions see http://reactivex.io

Asynchronous processing decouples I/O or computation from the thread that invoked the operation. A handle to the result is given back, usually a java.util.concurrent.Future or similar, that returns either a single object, a collection or an exception. Retrieving a result, that was fetched asynchronously is usually not the end of processing one flow. Once data is obtained, further requests can be issued, either always or conditionally. With Java 8 or the Promise pattern, linear chaining of futures can be set up so that subsequent asynchronous requests are issued. Once conditional processing is needed, the asynchronous flow has to be interrupted and synchronized. While this approach is possible, it does not fully utilize the advantage of asynchronous processing.

In contrast to the preceding examples, Observable objects answer the multiplicity and asynchronous questions in a different fashion: By inverting the Pull pattern into a Push pattern.

An Observable is the asynchronous/push “dual” to the synchronous/pull Iterable

An Observable<T> supports emission sequences of values or even infinite streams, not just the emission of single scalar values (as Futures do), which means an Observable<T> can emit 0 to N events. You will very much appreciate this fact once you start to work on streams instead of single values.

An Observable<T> is not biased toward some particular source of concurrency or asynchronicity and how the underlying code is executed - synchronous or asynchronous, running within a ThreadPool. As a consumer of an Observable<T>, you leave the actual implementation to the supplier, who can change it later on without you having to adapt your code.

The last key point of an Observable<T> is that the underlying processing is not started at the time the Observable<T> is obtained, rather its started at the moment an observer subscribes to the Observable<T>. This is a crucial difference to a java.util.concurrent.Future, which is started somewhere at the time it is created/obtained. So if no observer ever subscribes to the Observable<T>, nothing ever will happen.

All commands return an Observable<T> to which an Observer can subscribe to. That Observer reacts to whatever item or sequence of items the Observable<T> emits. This pattern facilitates concurrent operations because it does not need to block while waiting for the Observable to emit objects. Instead, it creates a sentry in the form of an Observer that stands ready to react appropriately at whatever future time the Observable does so.

The first thing you want to do when working with observables is to consume them. Consuming an observable means subscribing to it. Here is an example that subscribes and prints out all the items emitted:

The example prints the following lines:

You can see that the Subscriber (or Observer) gets notified of every event and also receives the completed event. An Observable<T> emits items until either an exception is raised or the Observable<T> finishes the emission calling onCompleted. No further elements are emitted after that time.

A call to the subscribe method returns a Subscription that allows to unsubscribe and, therefore, do not receive further events. Observables can interoperate with the un-subscription and free resources once a subscriber unsubscribed from the Observable<T>.

You can control the elements that are processed by your Subscriber using operators. The take() operator limits the number of emitted items if you are interested in the first N elements only.

The example prints the following lines:

Note that the take operator implicitly unsubscribes from the Observable<T> once the expected count of elements was emitted.

A subscription to an Observable<T> can be done either by another Observable or a Subscriber. Unless you are implementing a custom Observer, always use Subscriber. The used subscriber Action1 from the example above does not handle Exceptions so once an Exception is thrown you will see a stack trace like this:

It is always recommended to implement an error handler right from the beginning. At a certain point, things can and will go wrong.

A fully implemented subscriber declares the onCompleted and onError methods allowing you to react on these events:

The examples from above illustrated how observables can be set up in a not-opinionated style about blocking or non-blocking execution. An Observable<T> can be converted explicitly into a BlockingObservable<T>, which then behaves very much like an Iterable<T>.

The example prints the following line:

A blocking observable can be used to synchronize the observable chain and find back a way into the plain and well-known Pull pattern.

The toList operator collects all emitted elements and passes the list through the BlockingObservable<T>.

The example prints the following line:

There are many ways to establish observables. You have already seen just(), take() and toList(). Refer to the RxJava documentation for many more methods that you can use to create observables.

lettuce observables can be used for initial and chaining operations. When using lettuce observables, you will notice the non-blocking behavior. This is because all I/O and command processing are handled asynchronously using the netty EventLoop.

lettuce exposes its observables on the Standalone, Sentinel, Publish/Subscribe and Cluster APIs.

Connecting to Redis is insanely simple:

In the next step, obtaining a value from a key requires the GET operation:

Alternatively, written in Java 8 lambdas:

The execution is handled asynchronously, and the invoking Thread can be used to processed in processing while the operation is completed on the Netty EventLoop threads. Due its decoupled nature, the calling method can be left before the execution of the Observable<T> is finished. You need to synchronize the execution of the Observable<T> if you wish to wait for the Observable<T> to finish.

lettuce observables can be used within the context of observable chaining to load multiple keys asynchronously:

Alternatively, written in Java 8 lambdas:

There is a distinction between observables that was not covered yet:

• A cold Observable waits for a subscription until it emits values and does this freshly for every subscriber.

• A hot Observable begins emitting values upfront and presents them to every subscriber subsequently.

All Observables returned from the Redis Standalone, Redis Cluster, and Redis Sentinel API are cold, meaning that no I/O happens until they are subscribed to. As such an observer is guaranteed to see the whole sequence from the beginning. So just creating an Observable will not cause any network I/O thus creating and discarding Observables is cheap. Observables created for a Publish/Subscribe emit PatternMessages and ChannelMessages once they are subscribed to. Observables guarantee however to emit all items from the beginning until their end. While this is true for Publish/Subscribe observables, the nature of subscribing to a Channel/Pattern allows missed messages due to its subscription nature and less to the Hot/Cold distinction of observables.

Observables can transform the emitted values in various ways. One of the most basic transformations is flatMap() which you have seen from the examples above that converts the incoming value into a different one. Another one is map(). The difference between map() and flatMap() is that flatMap() allows you to do those transformations with Observable<T> calls.

The first flatMap() function is used to retrieve a value and the second flatMap() function appends the value to a Redis list named result. The flatMap() function returns an Observable whereas the normal map just returns <T>. You will use flatMap() a lot when dealing with flows like this, you’ll become good friends.

An aggregation of values can be achieved using the scan() transformation. It applies a function to each value emitted by an Observable<T>, sequentially and emits each successive value. We can use it to aggregate values, to count the number of elements in multiple Redis sets:

The aggregation function of scan() is applied on each emitted value, so three times in the example above. If you want to get the last value, which denotes the final result containing the number of elements in all Redis sets, apply the last() transformation:

Now let’s take a look at grouping observables. The following example emits three observables and groups them by the beginning character.

Alternatively, written in Java 8 lambdas:

The example prints the following lines:

The presence and absence of values is an essential part of reactive programming. Traditional approaches consider null as an absence of a particular value. With Java 8, Optional<T> was introduced to encapsulate nullability.

In the scope of Redis, an absent value is an empty list, a non-existent key or any other empty data structure. Reactive programming discourages the use of null as value. Newer reactive specifications, like reactive-streams.org prohibit the use of null. The reactive answer to absent values is just not emitting any value that is possible due the 0 to N nature of Observable<T>.

Suppose we have the keys Ben and Michael set each to the value value. We query those and another, absent key with the following code:

The example prints the following lines:

The output is just two values. The GET to the absent key Mark does not emit a value.

The reactive API provides operators to work with empty results when you require a value. You can use one of the following operators:

• defaultIfEmpty: Emit a default value if the Observable<T> did not emit any value at all

• switchIfEmpty: Switch to a fallback Observable<T> to emit values

• isEmpty: Emit an Observable<Boolean> that contains a flag whether the original Observable<T> is empty

• firstOrDefault/singleOrDefault/lastOrDefault/elementAtOrDefault: Positional operators to retrieve the first/last/Nth element or emit a default value

The values emitted by an Observable<T> can be filtered in case you need only specific results. Filtering does not change the emitted values itself. Filters affect how many items and at which point (and if at all) they are emitted.

The code will fetch only the keys Michael and Mark but not Ben. The filter criteria are whether the key starts with a M.

You already met the last() filter to retrieve the last value:

the extended variant of last() allows you to take the last N values:

The example from above takes the last 2 values.

The opposite to last() is the first() filter that is used to retrieve the first value:

Error handling is an indispensable component of every real world application and should to be considered from the beginning on. RxJava provides several mechanisms to deal with errors.

In general, you want to react in the following ways:

• Return a default value instead

• Use a backup observable

• Retry the observable (immediately or with delay)

The following code falls back to a default value after it throws an exception at the first emitted item:

You can use a backup Observable<T> which will be called if the first one fails.

It is possible to retry the observable by re-subscribing. Re-subscribing can be done as soon as possible, or with a wait interval, which is preferred when external resources are involved.

Use the following code if you want to retry with backoff:

Alternatively, written in Java 8 lambdas:

The attempts get passed into the retryWhen() method and zipped with the number of seconds to wait. The timer method is used to complete once its timer is done.

Schedulers in RxJava are used to instruct multi-threading. Some operators have variants that take a Scheduler as a parameter. These instruct the operator to do some or all of its work on a particular Scheduler.

RxJava ships with a set of preconfigured Schedulers, which are all accessible through the Schedulers class:

• Schedulers.computation(): Executes the computational work such as event-loops and callback processing.

• Schedulers.immediate(): Executes the work immediately in the current thread

• Schedulers.io(): Executes the I/O-bound work such as asynchronous performance of blocking I/O, this scheduler is backed by a thread-pool that will grow as needed

• Schedulers.newThread(): Executes the work on a new thread

• Schedulers.trampoline(): Queues work to begin on the current thread after any already-queued work

• Schedulers.from(): Create a scheduler from a java.util.concurrent.Executor

• Schedulers.test(): Test scheduler that allows you to exercise fine-tuned manual control over how the Scheduler’s clock behaves.

Do not use the computational scheduler for I/O.

Observables can be executed on a scheduler in the following different ways:

• Using an operator that makes use of a scheduler

• Explicitly by passing the Scheduler to such an operator

• By using subscribeOn(Scheduler)

• By using observeOn(Scheduler)

Operators like buffer, replay, skip, delay, parallel, and so forth use a Scheduler by default if not instructed otherwise. A list of default Schedulers for RxJava Observable Operators can be found here

All of the listed operators allow you to pass in a custom scheduler if needed. Sticking most of the time with the defaults is a good idea.

If you want the subscribe chain to be executed on a specific scheduler, you use the subscribeOn() operator. The code is executed on the main thread without a scheduler set:

The example prints the following lines:

This example shows the subscribeOn() method added to the flow (it does not matter where you add it):

The output of the example shows the effect of subscribeOn(). You can see that the Observable is executed on the same thread, but on the computation thread pool:

If you apply the same code to lettuce, you’ll notice a small difference:

The example prints the following lines:

Two things differ from the standalone examples:

1. The values are set rather concurrently than sequentially

2. The second flatMap() transformation prints the netty EventLoop thread name

This is because the lettuce observables are executed and completed on the netty EventLoop threads by default.

observeOn instructs an Observable to call its observer’s onNext, onError, and onCompleted methods on a particular Scheduler. Here, the order matters:

Everything before the observeOn() call is executed in main, everything below in the scheduler:

Schedulers allow direct scheduling of operations. Refer to the RxJava documentation for further information.

See Transactions.