RxJava - Reactive Programming

Reactive Manifesto

Required to follow the principles of the Reactive Manifesto.

• Responsive – systems should respond in a timely manner
• Message Driven – systems should use async message-passing between components to ensure loose coupling
• Elastic – systems should stay responsive under high load
• Resilient – systems should stay responsive when some components fail

Basics

• Observable represents any object that can get data from a data source and whose state may be of interest in a way that other objects may register an interest
• An observer is any object that wishes to be notified when the state of another object changes

Base Clases

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

Flowable

Flowable implements the Reactive-Streams Pattern, which is a standard for asynchronous stream processing with non-blocking backpressure.

• Support Observable emitting more rapidly than observer consume
• Operators: map,filter,reduce
• Schedulers
• Hot & Cold Flowables: lowable can be either "hot" (it can start emitting items as soon as it's created) or "cold" (it doesn't start emitting items until a subscriber starts to observe).

Difference between map and reduce

map()

The map operator is used to transform the items emitted by an Observable by applying a function to each item. The function transforms the item, and the transformed item is then emitted by the Observable.

Flowable.range(1, 5)
    .map(v -> v * 2)
    .subscribe(System.out::println);

reduce()

The reduce operator is used to combine all items emitted by an Observable into a single item by applying a function that takes two parameters: the accumulated value and the current item. The function returns a new accumulated value, which is then used in the next call to the function.

Flowable.range(1, 5)
    .reduce(0, (total, next) -> total + next)
    .subscribe(System.out::println);

Schedulers

Schedulers in RxJava are used to specify where and when tasks are executed, whether it be on a new thread, a pooled thread, or the current thread.

Schedulers.newThread()

This creates a new thread for each unit of work scheduled.

Flowable.range(1, 5)
.subscribeOn(Schedulers.newThread())
.map(v -> v * v)
.blockingSubscribe(System.out::println);

Schedulers.io()

This maintains a pool of threads for IO-bound work. In this example, the work is done on one of the pooled threads. This is useful for IO-bound work like reading and writing to disk, network calls, database queries, etc.

Flowable.range(1, 5)
    .subscribeOn(Schedulers.io())
    .map(v -> v * v)
    .blockingSubscribe(System.out::println);

Schedulers.computation()

This maintains a fixed number of threads based on the number of available processors. It's intended for computational work. In this example, the work is done on one of the computation threads. This is useful for CPU-intensive work like calculations and transformations.

Flowable.range(1, 5)
.subscribeOn(Schedulers.computation())
.map(v -> v * v)
.blockingSubscribe(System.out::println);

Schedulers.single()

This executes all tasks on a single thread.

Flowable.range(1, 5)
.subscribeOn(Schedulers.single())
.map(v -> v * v)
.blockingSubscribe(System.out::println);

Schedulers.trampoline()

This executes tasks in a FIFO manner on the immediate scheduler.

Flowable.range(1, 5)
.subscribeOn(Schedulers.trampoline())
.map(v -> v * v)
.blockingSubscribe(System.out::println);

Benchmark