Skip to content

A Guide to Multithreading in RxJava (and observeOn vs subscribeOn)

Following the finale of the Knowing your RxJava Operator Toolbelt series, you’d think that we’ve covered a huge part of the RxJava ecosystem, but well eurghuehue.

C’est wrong!

RxJava’s operators are only the surface of the library. Let’s go back to the roots of how RxJava describes itself on its website:

A library for composing asynchronous and event-based programs using observable sequences for the Java VM.

Asynchronous… Hmmmmmm…….

When do we use asynchronous tasks? We’d use them when we don’t want to clog the main thread with heavy lifting, right? But by default, RxJava doesn’t do that. By default, observables are created on the main thread.

Now if you’ve read about Utility Operators, you may have noticed two little operators that did something a bit different from the others. They change the thread which the observable runs on. In general, this practice is called multithreading, and as a quick reminder, taken directly from the page itself, the operators are:

——–

observeOn

Observable.just(1, 2, 3) // UI THREAD
        .map { i -> i + 1 } // UI THREAD
        .observeOn(Schedulers.io()) // BACKGROUND THREAD
        .map { i -> i * 2 } // BACKGROUND THREAD
        .subscribe() // BACKGROUND THREAD

One of the most commonly used operators, and for good reasons. This allows you to specify the thread for which your other operators run from.

I could write a whole article on threading, but as a starter for example, you can choose to observe on the background thread (Schedulers.io) if you have an observable with beefy map functions and you don’t want to freeze the UI which runs on the main thread.

subscribeOn

Observable.just(1, 2, 3) // BACKGROUND THREAD
        .map { i -> i + 1 } // BACKGROUND THREAD
        .subscribeOn(Schedulers.io()) // BACKGROUND THREAD
        .map { i -> i * 2 } // BACKGROUND THREAD
        .subscribe() // BACKGROUND THREAD

This allows you to specify the thread for which your other operators run from… as well!

Okay the difference between this and ObserveOn is the operators they affect. ObserveOn will only affect the operators that come after it, while SubscribeOn will affect the whole observable.

———

As you can probably see, switching between the threads is piss-easy. What you need to know is which threads to switch to so that you can ensure your app runs as smoothly as possible.

Schedulers (or Threads)

Scheduler is the name we give to the things that assign tasks to threads, but to make it easier, we can view them as the threads themselves. This is what we pass into subscribeOn or ObserveOn.

AndroidScheulders.mainThread

This is quite clearly, the main thread, and the only scheduler on the main thread. We put all UI operations here because, well, your app would crash otherwise.

implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

This however, isn’t available with the base RxJava library, but rather with RxAndroid.

Scheulders.io

Quite easily the most commonly used background thread as it is backed by an unbounded thread pool. This means the size of its thread pool can grow as needed, but don’t get carried away. You don’t want to flood the thread pool or your app won’t be performing well. IO is used for non-CPU intensive IO tasks like accessing the database, performing network calls, and accessing the file system

Schedulers.computation

Like a contrary to IO, computation is backed by a bounded thread pool. Its size corresponds to the number of processors available on the device. Computation is used for CPU-intensive tasks like image manipulation and processing large data sets.

What’s a Thread Pool?

If you’re confused, don’t worry. Imagine Thread Pool as a queue, waiting for tasks to occupy its thread. When we talk about the size of a thread pool, you can think of it as the number of separate lines in the queue, like having more tills at a grocery store. With IO, you can have an infinite number of these ’tills’, but the more you have, the weaker these tills will operate. Computation limits these so that its threads will be given more processing power to handle intensive tasks.

Schedulers.newThread

NewThread doesn’t make use of any thread pool but instead creates its own thread separate from everything else. As creating and destroying is a heavy task for the processors, you should use this very sparingly. NewThread is suitable if you want to create long-running isolated background tasks.

Schedulers.single

Single is a single long thread pool used to execute tasks on a First-In-First-Out basis. Use it when you have a series of tasks from different parts of your app that you can’t afford being performed together.

Schedulers.trampoline

Trampoline schedules the task on the current thread (where the code is being executed), effectively blocking it. However, if there is another task still being processed, trampoline will wait for that task to finish before executing the task. Use this if you have a task that by every means needs to be executed immediately, and since it blocks the thread, only use it if you really must.

Schedulers.from

This takes in an executor (java.util.concurrent) to transform it into a scheduler for use with observables. In a way, you can see this as a custom scheduler Admittedly, I’m not very familiar with executors myself, but this seems to be a good place to get started if it interests you.

Tags: