RxJava: Reactive Extensions for the JVM

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

Version 4.x (Javadoc)

:+1: Native Java 25* implementation. Will go 26 or even 27 depending on how long it takes.
:+1: No 3rd party library required at runtime.
:+1: JPMS and :question: OSGi support still intact.
:+1: java.util.concurrent.Flow-based implementation.
:+1: Virtual Thread support; virtualCreate(), virtualTransform(), :eye: Schedulers.virtual().
:information_source: Reactive Streams Test Compatibility Kit usage; Reactive-Streams.
:satellite: Rewamp of the javadoc bloat in the base types via sealed interfaces.
:satellite: Reduce overload bloat by using record-based configurations.
:satellite: Internal optimizations now that I have the master :key:.
:eye: Possible usages for Scoped variables for context and per-item resource management.
:eye: Possible use for the Java Cleaner API.
:eye: Possible new monad Streamable<T> built around Virtual Threads & virtual blocking. Think IAsyncEnumerable for Java.
:eye: Possible inclusion of 2nd and 3rd party operators.
:eye: Possible inclusion of the Iterable Extensions (Ix) 2nd party library. ju.Stream is sh|t wrt interfacing and composability.
:question: Android compatibility depends on your API level and what desugaring is available.
:lady_beetle: Resolve many anomalies and bugs with operators such as groupBy, window, concat, etc.
:warning: RxJava 3.x support will be toned down in the coming months, will be offered for +1 year after 4.x official release.

Getting started

Setting up the dependency

The first step is to include RxJava 4 into your project, for example, as a Gradle compile dependency:

implementation "io.reactivex.rxjava4:rxjava:4.x.y"

(Please replace x and y with the latest version numbers: )

Hello World

The second is to write the Hello World program:

package rxjava.examples;

import io.reactivex.rxjava4.core.*;

public class HelloWorld {
    public static void main(String[] args) {
        Flowable.just("Hello world").subscribe(System.out::println);
    }
}

Note that RxJava 4 components now live under io.reactivex.rxjava4 and the base classes and interfaces live under io.reactivex.rxjava4.core.

Base classes

RxJava 4 features several base classes you can discover operators on:

io.reactivex.rxjava4.core.Flowable: 0..N flows, supporting Reactive-Streams and backpressure
io.reactivex.rxjava4.core.Observable: 0..N flows, no backpressure,
io.reactivex.rxjava4.core.Single: a flow of exactly 1 item or an error,
io.reactivex.rxjava4.core.Completable: a flow without items but only a completion or error signal,
io.reactivex.rxjava4.core.Maybe: a flow with no items, exactly one item or an error.

Some terminology

Upstream, downstream

The dataflows in RxJava consist of a source, zero or more intermediate steps followed by a data consumer or combinator step (where the step is responsible to consume the dataflow by some means):

source.operator1().operator2().operator3().subscribe(consumer);

source.flatMap(value -> source.operator1().operator2().operator3());

Here, if we imagine ourselves on operator2, looking to the left towards the source is called the upstream. Looking to the right towards the subscriber/consumer is called the downstream. This is often more apparent when each element is written on a separate line:

source
  .operator1()
  .operator2()
  .operator3()
  .subscribe(consumer)

Objects in motion

In RxJava's documentation, emission, emits, item, event, signal, data and message are considered synonyms and represent the object traveling along the dataflow.

Backpressure

When the dataflow runs through asynchronous steps, each step may perform different things with different speed. To avoid overwhelming such steps, which usually would manifest itself as increased memory usage due to temporary buffering or the need for skipping/dropping data, so-called backpressure is applied, which is a form of flow control where the steps can express how many items are they ready to process. This allows constraining the memory usage of the dataflows in situations where there is generally no way for a step to know how many items the upstream will send to it.

In RxJava, the dedicated Flowable class is designated to support backpressure and Observable is dedicated to the non-backpressured operations (short sequences, GUI interactions, etc.). The other types, Single, Maybe and Completable don't support backpressure nor should they; there is always room to store one item temporarily.

Assembly time

The preparation of dataflows by applying various intermediate operators happens in the so-called assembly time:

Flowable<Integer> flow = Flowable.range(1, 5)
.map(v -> v * v)
.filter(v -> v % 3 == 0)
;

At this point, the data is not flowing yet and no side-effects are happening.

Subscription time

This is a temporary state when subscribe() is called on a flow that establishes the chain of processing steps internally:

flow.subscribe(System.out::println)

This is when the subscription side-effects are triggered (see doOnSubscribe). Some sources block or start emitting items right away in this state.

Runtime

This is the state when the flows are actively emitting items, errors or completion signals:


Observable.create(emitter -> {
     while (!emitter.isDisposed()) {
         long time = System.currentTimeMillis();
         emitter.onNext(time);
         if (time % 2 != 0) {
             emitter.onError(new IllegalStateException("Odd millisecond!"));
             break;
         }
     }
})
.subscribe(System.out::println, Throwable::printStackTrace);

Practically, this is when the body of the given example above executes.

Simple background computation

One of the common use cases for RxJava is to run some computation, network request on a background thread and show the results (or error) on the UI thread:

import io.reactivex.rxjava4.schedulers.Schedulers;

Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
})
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.single())
  .subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000); // <--- wait for the flow to finish

This style of chaining methods is called a fluent API which resembles the builder pattern. However, RxJava's reactive types are immutable; each of the method calls returns a new Flowable with added behavior. To illustrate, the example can be rewritten as follows:

Flowable<String> source = Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
});

Flowable<String> runBackground = source.subscribeOn(Schedulers.io());

Flowable<String> showForeground = runBackground.observeOn(Schedulers.single());

showForeground.subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000);

Typically, you can move computations or blocking IO to some other thread via subscribeOn. Once the data is ready, you can make sure they get processed on the foreground or GUI thread via observeOn.

Schedulers

RxJava operators don't work with Threads or ExecutorServices directly but with so-called Schedulers that abstract away sources of concurrency behind a uniform API. RxJava 4 features several standard schedulers accessible via Schedulers utility class.

Schedulers.computation(): Run computation intensive work on a fixed number of dedicated threads in the background. Most asynchronous operators use this as their default Scheduler.
Schedulers.io(): Run I/O-like or blocking operations on a dynamically changing set of threads.
Schedulers.single(): Run work on a single thread in a sequential and FIFO manner.
Schedulers.trampoline(): Run work in a sequential and FIFO manner in one of the participating threads, usually for testing purposes.

These are availab

RxJava

Install / Use

README