13. Intro to Schedulers

Written by Alex Sullivan & Junior Bontognali

Until now, you’ve managed to work with schedulers, while avoiding any explanation about how they handle threading or concurrency. In earlier chapters, you used methods, which implicitly used some sort of concurrency/threading level, such as the buffer or interval operators.

You probably have a feeling that schedulers have some sort of magic under the hood, but before you understand schedulers, you’ll also need to understand what those observeOn and subscribeOn functions are all about.

This chapter is going to cover the beauty behind schedulers. You’ll learn why the Rx abstraction is so powerful and why working with asynchronous programming is far less painful than using AsyncTasks, IntentHandlers and the myriad of other asynchronous tools Android development offers.

Note: Creating custom schedulers is beyond of the scope of this book. Keep in mind that the schedulers and initializers provided by RxJava generally cover 99% of cases. Always try to use the built-in schedulers.

What is a scheduler?

Before getting your hands dirty with schedulers, it’s important to understand what they are — and what they are not. To summarize, a scheduler is an abstraction introduced by the RxJava library to schedule work at some point in time. The work happens in some asynchronous context. That context could be custom Threads, an event loop, Executors and so on.

While the Scheduler abstract class is a powerful abstraction over different ways of executing asynchronous code, for Android apps you can usually think of schedulers in relation to threads and thread pools. You’ll learn more about the different types of schedulers and how they allow you to switch between threading contexts later on.

Here’s a good example as to how schedulers can be used:

In this diagram, you have the concept of a cache operator. An Observable makes a request to a server and retrieves some data. This data is processed by a custom operator named cache, which stores the data somewhere. After this, the data is passed to all subscribers in a different scheduler, most likely the main scheduler, which sits on top of the Android main thread. Remember that anytime you update a UI element in an Android app it must be done on the main thread.

Setting up the project

Time to write some code! In this project, you are going to work with an Android app called Schedulers that has a profoundly beautiful user interface. That user interface is one TextView in the center of a white screen.

You’ll work on this project in Android Studio instead of IntelliJ IDEA, because, in this chapter, you’ll also be introduced to the RxAndroid library, which requires Android dependencies.

To gaze upon the beginnings of this magnificent app, Open Android studio to its initial screen and select “Open existing project”:

This will cause Android studio to load up and build the project. When it finishes (be patient), you’ll see a Play button appear in the top toolbar next to the connected device (or emulator), that’s after a litte Android icon and the word app:

Now, use the Play button in the top toolbar to build and run the app. You’ll see a very basic interface when it runs:

While you’ll technically be working on an Android app, you’ll be focused almost entirely on the Logcat output, which you can find in the bottom console of Android Studio:

Logcat can get pretty noisy, so you should make sure that you’re filtering the output to the Schedulers app and filtering it further by including the main TAG used by the app when logging. The TAG is “SchedulerLogging” and you can filter based off that tag by adding the string in the search box at the top right of the Logcat window:

Inspect the filtered Logcat output, and you should see the following:

0s | [D] [dog] received on Thread: main
0s | [S] [dog] received on Thread: main

Before proceeding, open X.kt and take a look at the implementation of dump and dumpingSubscription.

The first method dumps the element and the current thread information inside a doOnNext operator using the [D] prefix. The second does the same using the [S] prefix, but calls subscribe. Both methods indicate the elapsed time, so the 0s above stand for “0 seconds elapsed.”

Switching schedulers

One of the most important things in Rx is the ability to switch schedulers at any time, without any restrictions except for ones imposed by the inner process generating events.

Note: An example of that type of restriction is if the Observable emits non-thread safe objects, which cannot be sent across threads. In that case, RxJava will allow you to switch schedulers, but you would be violating the logic of the underlying code.

To understand how schedulers behave, you’ll create a simple Observable to play with that provides some fruit.

Add the following code to the bottom of the onCreate method in SchedulersActivity.kt:

val fruit = Observable.create<String> { observer ->
  observer.onNext("[apple]")
  Thread.sleep(2000)
  observer.onNext("[pineapple]")
  Thread.sleep(2000)
  observer.onNext("[strawberry]")
}

This Observable features a Thread.sleep function. While this is not something you’d usually see in real apps, in this case, it will help you understand how subscriptions and observations work.

Add the following code to subscribe to the Observable you created:

fruit
  .dump()
  .dumpingSubscription()
  .addTo(disposables)

Build and run, and check out the logging in the console:

0s | [D] [dog] received on Thread: main
0s | [S] [dog] received on Thread: main
0s | [D] [apple] received on Thread: main
0s | [S] [apple] received on Thread: main
2s | [D] [pineapple] received on Thread: main
2s | [S] [pineapple] received on Thread: main
4s | [D] [strawberry] received on Thread: main
4s | [S] [strawberry] received on Thread: main

The starter project already contained code creating a animal behavior subject and subscribing and dumping the contents. So here you have the original subject, followed by a fruit every two seconds after that.

The fruit is generated on the main thread, but it would be nice to move it to a background thread. Growing fruit takes time after all, and you wouldn’t want to block your main thread while it’s growing! To create the fruit in a background thread, you’ll have to use subscribeOn.

Using subscribeOn

In some cases, you might want to change on which scheduler the Observable computation code runs — not the code in any of the subscription operators, but the code that is actually emitting the Observable events.

Note: For the custom Observable that you have created, the code that emits events is the one you supply as the trailing lambda for Observable.create { ... }.

The way to set the scheduler for that computation code is to use the subscribeOn operator. It might sound like a counterintuitive name at first glance, but after thinking about it for a while, it starts to make sense. When you want to actually observe an Observable, you subscribe to it. This determines where the original processing will happen. If subscribeOn is not called, then RxJava automatically uses the current thread:

This process is creating events on the main thread using the main scheduler. The AndroidSchedulers.mainThread() that you’ve used in previous chapters sits on top of the main thread. All the tasks you want to perform on the main thread have to use this scheduler, which is why you used it in previous examples when working with the UI. To switch schedulers, you’ll use subscribeOn.

As noted previously, the subscribeOn operator allows you to provide a Scheduler to change what thread the Observable creation code is called on. However, before you can use the operator, you need an instance of Scheduler.

RxJava provides a Schedulers (notice the trailing s in that class name) utility class that contains several instances of predefined schedulers, as well as a few utility methods to create new schedulers from existing Java concepts like Executor.

For this example, you’ll use the io scheduler that you’ve used in past projects. You’ll see a detailed breakdown of the different types of default schedulers you can use later on in the chapter.

To use the scheduler, replace the previous subscription to fruits you created with this new one:

fruit
  .subscribeOn(Schedulers.io())
  .dump()
  .dumpingSubscription()
  .addTo(disposables)

Now that your new scheduler is in place, build and run and check the result:

0s | [D] [dog] received on Thread: main
0s | [S] [dog] received on Thread: main
0s | [D] [apple] received on Thread: RxCachedThreadScheduler-1
0s | [S] [apple] received on Thread: RxCachedThreadScheduler-1
2s | [D] [pineapple] received on Thread: RxCachedThreadScheduler-1
2s | [S] [pineapple] received on Thread: RxCachedThreadScheduler-1
4s | [D] [strawberry] received on Thread: RxCachedThreadScheduler-1
4s | [S] [strawberry] received on Thread: RxCachedThreadScheduler-1

Under the hood, the Schedulers.io() method is returning a scheduler that works off of a thread pool. Those threads are cached and the library names them accordingly.

Now, both the Observable and the subscribed observer from the fruit Observable are processing data in the same thread.

Since you didn’t use the subscribeOn operator on the animal subscribing code, its objects are still being emitted on the main thread.

That’s cool, but what can you do if you want to change where the observer performs the code of your operators? You have to use observeOn.

Using observeOn

Observing is one of the three fundamental concepts of Rx. It involves an entity producing events and an observer for those events. In this case, and in opposition to subscribeOn, the operator observeOn changes the scheduler where the observation happens.

Once an event is pushed by an Observable to all the subscribed observers, this operator will ensure that the event is handled by the correct scheduler.

To switch from the io scheduler to the main thread, you need to call observeOn before subscribing.

There’s only one issue. RxJava has no idea what a main scheduler is. The animal subscription code is running on the main thread, but that’s just because RxJava defaults to using whatever thread calls the subscribing code if there’s no observeOn operator.

Remember, RxJava is a Java library that has no knowledge of Android. Since the main thread is specific to your Android app, you need some way of creating a scheduler that always routes work to the Android main thread.

You could write the logic yourself to wrap the Android main Looper in an RxJava scheduler. Luckily for you, someone else has already done that work!

Open the build.gradle file and add a new dependency for the RxAndroid library in the dependencies block:

implementation "io.reactivex.rxjava3:rxandroid:3.0.0"

RxAndroid is an extremely small library whose entire purpose is to expose the Android main looper as a scheduler via the AndroidSchedulers.mainThread() static utility function.

While the name of the library would imply that it interacts with all things Android, the maintainers of the library felt that it would be better to whittle the project down to only the most crucial element of using Rx on Android.

One more time, replace your fruits subscription code:

fruit
  .subscribeOn(Schedulers.io())
  .dump()
  .observeOn(AndroidSchedulers.mainThread())
  .dumpingSubscription()
  .addTo(disposables)

Run the project and check the Logcat output once more (you will need to wait a few seconds until the app stops printing):

0s | [D] [dog] received on Thread: main
0s | [S] [dog] received on Thread: main
0s | [D] [apple] received on Thread: RxCachedThreadScheduler-1
0s | [S] [apple] received on Thread: main
2s | [D] [pineapple] received on Thread: RxCachedThreadScheduler-1
2s | [S] [pineapple] received on Thread: main
4s | [D] [strawberry] received on Thread: RxCachedThreadScheduler-1
4s | [S] [strawberry] received on Thread: main

You’ve achieved the result you wanted: All the events are now processed on the correct thread. The Observable is processing and generating events on the background thread, and the subscribing observer is doing its job on the main thread.

This is a very common pattern: You often use a background process to retrieve data from a server and process the data received, only switching to the AndroidSchedulers.mainThread scheduler to process the final event and display the data in the user interface.

Pitfalls

The ability to switch schedulers and threads looks amazing, but it comes with some pitfalls. To see why, you’ll push some events to the subject using a new thread. Since you need to track on which thread the computation takes place, a good solution is to use Thread.

Right after the fruit Observable, add the following code to generate some animals:

val animalsThread = Thread {
  Thread.sleep(3000)
  animal.onNext("[cat]")
  Thread.sleep(3000)
  animal.onNext("[tiger]")
  Thread.sleep(3000)
  animal.onNext("[fox]")
  Thread.sleep(3000)
  animal.onNext("[leopard]")
}

Then name the thread, so you will be able to recognize it, and start it up:

animalsThread.name = "Animals Thread"
animalsThread.start()

Run the app. You should see your new thread in action:

...
3s | [D] [cat] received on Thread: Animals Thread
3s | [S] [cat] received on Thread: Animals Thread
4s | [D] [strawberry] received on Thread: RxCachedThreadScheduler-1
4s | [S] [strawberry] received on Thread: main
6s | [D] [tiger] received on Thread: Animals Thread
6s | [S] [tiger] received on Thread: Animals Thread
9s | [D] [fox] received on Thread: Animals Thread
9s | [S] [fox] received on Thread: Animals Thread
12s | [D] [leopard] received on Thread: Animals Thread
12s | [S] [leopard] received on Thread: Animals Thread

Perfect — you have animals created on the dedicated thread. Next, process the result on io scheduler.

Note: It might seem repetitive to keep adding code and then replacing it with something else, but the goal here is to compare the differences between the various schedulers.

Replace the original subscription to the animal subject with the following code:

animal
  .dump()
  .observeOn(Schedulers.io())
  .dumpingSubscription()
  .addTo(disposables)

Build and run, and the new result is as follows:

...
3s | [D] [cat] received on Thread: Animals Thread
3s | [S] [cat] received on Thread: RxCachedThreadScheduler-1
4s | [D] [strawberry] received on Thread: RxCachedThreadScheduler-2
4s | [S] [strawberry] received on Thread: main
6s | [D] [tiger] received on Thread: Animals Thread
6s | [S] [tiger] received on Thread: RxCachedThreadScheduler-1
9s | [D] [fox] received on Thread: Animals Thread
9s | [S] [fox] received on Thread: RxCachedThreadScheduler-1
12s | [D] [leopard] received on Thread: Animals Thread
12s | [S] [leopard] received on Thread: RxCachedThreadScheduler-1

Now you’re switching threads from the animals thread where the items are actually pushed to the subject, to one of the cached io threads provided by the Schedulers.io function.

What if you want the observation process on the io scheduler, but you want to handle the subscription on the main thread? For the first case, the observeOn is already correct, but for the second, it’s necessary to use subscribeOn.

Replace the animal subscription, this time with the following:

animal
  .subscribeOn(AndroidSchedulers.mainThread())
  .dump()
  .observeOn(Schedulers.io())
  .dumpingSubscription()
  .addTo(disposables)

Build and run, and you’ll get the following result:

3s | [D] [cat] received on Thread: Animals Thread
3s | [S] [cat] received on Thread: RxCachedThreadScheduler-2
4s | [D] [strawberry] received on Thread: RxCachedThreadScheduler-1
4s | [S] [strawberry] received on Thread: main
6s | [D] [tiger] received on Thread: Animals Thread
6s | [S] [tiger] received on Thread: RxCachedThreadScheduler-2
9s | [D] [fox] received on Thread: Animals Thread
9s | [S] [fox] received on Thread: RxCachedThreadScheduler-2
12s | [D] [leopard] received on Thread: Animals Thread
12s | [S] [leopard] received on Thread: RxCachedThreadScheduler-2

Wait?! What? Why isn’t the computation happening on the correct scheduler? Since you’re using the subscribeOn operator, you should be seeing the items being computed on the main scheduler, right? This is a common and dangerous pitfall that comes from thinking of Rx as asynchronous or multi-threaded by default — which isn’t the case.

Rx and the general abstraction is free-threaded; there’s no magic thread switching taking place when processing data. The computation is always performed on the original thread if not specified otherwise.

Note: Any thread switching happens after an explicit request by the programmer using the operators subscribeOn and observeOn.

Thinking Rx does some thread handling by default is a common trap to fall into. What’s happening above is a misuse of the Subject. The original computation is happening on a specific thread, and those events are pushed in that thread using Thread() { ... }. Due to the nature of Subject, Rx has no ability to switch the original computation scheduler and move to another thread, since there’s no direct control over where the subject is pushed.

Why does this work with the fruit thread though? That’s because using Observable.create puts Rx in control of what happens inside the Thread block so that you can more finely customize thread handling.

This unexpected outcome is commonly known as the Hot and Cold Observables problem.

In the case above, you are dealing with a hot Observable. The Observable doesn’t have any side-effect during subscription, but it does have its own context in which events are generated and RxJava can’t control it (namely, it sports its own Thread).

A cold Observable in contrast doesn’t produce any elements before any observers subscribe to it. That effectively means it doesn’t have its own context until, upon subscription, it creates some context and starts producing elements.

Hot vs. cold

The section above touched on the topic of hot and cold Observables. The topic of hot and cold Observables is quite opinionated and generates a lot of debate, so let’s briefly look into it, here. The concept can be reduced to a very simple question:

Some examples of side effects are:

• Fire a request to the server
• Edit the local database
• Write to the file system
• Launch a rocket

The world of side effects is endless, so you need to determine whether your Observable instance is performing side effects upon subscription. If you can’t be certain about that, then perform more analysis or dig further into the source code. Launching a rocket on every subscription might not be what you’re looking to achieve…

Another common way to describe this is to ask whether or not the Observable shares side-effects. If you’re performing side effects upon subscription, it means that the side effect is not shared. Otherwise, the side effects are shared with all subscribers.

This is a fairly general rule, and applies to any RxJava object like a subject and related subtypes.

As you’ve certainly noticed, we haven’t spoken much about hot and cold Observables so far in the book. It’s a common topic in reactive programming, but in Rx you encounter the concept only in specific cases like the Thread example above or when you need greater control, such as when you run tests.

Keep this section as a point of reference, so in case you need to approach a problem in terms of hot or cold Observables, you can quickly open the book to this point and refresh yourself on the concept.

Best practices and built-in schedulers

Schedulers are a non-trivial topic, so they come with some best practices for the most common use cases. In this section, you’ll get a quick introduction to serial and concurrent schedulers, learn how they process the data and see which type works better for a particular context.

Android main scheduler

AndroidSchedulers.mainThread() sits on top of the main thread. This scheduler is used to process changes on the user interface and perform other high-priority tasks.

As a general practice when developing applications on Android, long-running tasks should not be performed using this scheduler, so avoid things like database requests or other heavy tasks. If you try and execute a network request from this scheduler you’ll receive a NetworkOnMainThreadException.

Additionally, if you perform side effects that update the UI, you must switch to this scheduler to make sure all UI updating logic happens on the main thread. If you don’t, you may see exceptions about modifying UI code from a different thread.

io scheduler

The scheduler returned by Schedulers.io() should be used whenever you’re doing work that’s IO bound. Specifically, if you’re making any network calls, accessing items from a database, or reading lines from a file, this is the scheduler for you.

Under the hood, it’s backed by thread pool that will grow as needed, so make sure not to do strict computational work while using the IO scheduler.

Computation scheduler

If you do need to heavy computational work, like crunching large data sets or handling event loops, you can use the scheduler returned by Schedulers.computation().

In opposition to the IO scheduler, the computation scheduler will not spawn more threads as needed. Instead, the number of threads it works with is normally limited to the number of cores the CPU has.

If you think about it this makes sense: If you’re doing computationally heavy work and you have more threads than number of cores in the CPU, you won’t be able to process the work any faster since all cores are occupied. Instead, you’d just be creating more memory overhead by creating new threads.

Single threaded scheduler

Sometimes, you need to work off the main thread but you also need guarantees that the work you’re doing is happening sequentially. This isn’t a problem if you’re only working in the confines of one RxJava chain, since, for the most part, those chains will always happen sequentially.

However, if you have multiple distinct chains and you want to know that you’re continually adding new work to a queue, you can use the Schedulers.single scheduler.

The single scheduler is potentially the simplest of all the schedulers. It’s ultimately backed by one thread. That means that, whenever you queue up new work on that thread, it’s queued to the bottom so you know it happens after other work you’ve added before.

Trampoline scheduler

Similar to the single scheduler, the scheduler returned by Schedulers.trampoline() always operates on a single thread. Unlike the single scheduler, that thread isn’t a background thread. Instead, it’s the main thread that created the trampoline scheduler. You’ll see in Chapter 15, “Testing RxJava Code,” that the trampoline scheduler can be very useful while writing unit tests.

Test scheduler

TestScheduler is a special kind of beast. It’s meant only to be used in testing, so make sure not to use this scheduler in production code. This special scheduler simplifies operator testing. You will have a look into using this scheduler in the dedicated chapter about testing, but let’s have a quick look since you’re doing the grand tour of schedulers.

Open the SchedulerTest.kt file. It’s a simple unit test that attempts to test the Observable.timer method. As any good developer knows, testing code that interacts with time can be extremely challenging. Without TestScheduler, you may be forced to block the test from finishing until a certain amount of time has passed. That makes for very slow unreliably tests, which is a big no-no in the testing world.

TestScheduler allows you to control how much “time” has passed and how actions and events are triggered.

Take a look at the following code:

val scheduler = TestScheduler()
val observable = Observable.timer(2, TimeUnit.SECONDS, scheduler)

You’re creating an instance of TestScheduler and then passing that schedule in to the Observable.timer method. You have seen the Observable.timer factory method, but you may not have used the version that takes in a scheduler yet.

Command-click into the Observable.timer method and scroll up one method signature to the version of Timer that doesn’t take a scheduler. In that methods JavaDocs, you’ll see the following:

*  <dd>{@code timer} operates by default on the {@code computation} {@link Scheduler}.</dd>

By default, most timing oriented operators will operate on the computation scheduler you saw earlier. That can create problems if you want to test the code later on or if you expect the code to be run on whatever thread it was started on.

Going back to the example unit test, you see the following code:

val testTimer = observable.test()

testTimer.assertNotComplete()

scheduler.advanceTimeBy(2, TimeUnit.SECONDS)

testTimer.assertComplete()

You’ll learn about the test method on Observables in Chapter 15, “Testing RxJava Code.” All you need to know for now is thatu it allows you to assert certain events have happened on yor Observable. In the above example you’re first asserting that the Observable has not completed yet, which makes sense because the Observable only completes after two seconds.

Then you’re using the TestScheduler.advanceTimeBy method to artificially advance what that scheduler thinks of as the current time. Kind of like time traveling, except it makes for a far less interesting sci-fi television series.

Since you’ve advanced time by two seconds, that means the Observable should have emitted its value and completed.

Sure enough, if you run the unit test by clicking the small green arrow next to the test method, you’ll see that it passed.

You’ll learn more about how amazing TestScheduler is later on.

Key points

• A Scheduler is an abstract context upon which RxJava executes work. In other words, Schedulers let you choose to do work on different threads.
• You can use the subscribeOn operator to control on what thread your Observable is created. That allows you to, for example, execute the actual networking portion of an API call off the main thread.
• After using subscribeOn, you can use the observeOn operator to then choose a different thread to actually receive the emitted objects on. You’ll often use this operator to switch back to the main thread to update UI objects.
• While subscribeOn and observeOn are extremely powerful operators, they’re not magic. If you call the onNext method of a subject on a different thread, RxJava can’t honor your subscribeOn call and you’ll see the item emitted on the original thread.
• There are both hot Observables and cold Observables. cold Observables create some special side effect when they’re subscribed to. A network call that returns an Oobservable is an example of a cold stream. A hot Observable is always running and emitting items, even if no one is listening. Subscribing to a hot Observable will not cause any special side effects.
• There are several built in schedulers for you to use. The io scheduler is great for network and database calls, while the computation scheduler is good for event loops and computationally expensive code.
• The RxAndroid library exposes another special scheduler you can use to emit items on the Android main thread.
• Finally, the TestScheduler class assists in testing RxJava code and should not be used in production code.

Where to go from here?

Schedulers are a non-trivial topic in the Rx space; they’re responsible for computing and performing all tasks in RxJava.

Before proceeding, invest some time in playing around with the examples in this chapter and test some schedulers to see what impact they have on the final result. Understanding schedulers will make life easier with RxJava, and will improve your confidence when using subscribeOn and observeOn.