As we know, one of RxJava’s important features is that we can easily manage threads by using subscribeOn and observeOn operators. In fact, it's been a go-to solution for handling background operations on Android for some time now. However, this use case of RxJava comes with an easy to miss caveat: sometimes subscribeOn doesn't work as you expect! Let's see when and why on a couple of examples.

Observable#create and emitters

When using RxJava we can pass a scheduler to a subscribeOn operator, which will cause the subscription to run on a thread provided by that scheduler. Normally the subscription will emit items on the same thread it's called. But nothing prevents us from emitting items from a different thread! Let's try it:

And the result:

[RxComputationThreadPool-1] Subscribing

[Main thread] 1 - emitting

[Main thread] 1 - performing an expensive computation

[Main thread] 2 - emitting

[Main thread] 2 - performing an expensive computation

[Main thread] 3 - emitting

[Main thread] 3 - performing an expensive computation

As we see in the output, both the emission and the mapping are performed on the newly created thread, and not one from the computation scheduler. This means that yes, RxJava will manage threading, but it will do it only as long as you don't change the threads manually.

In this case, the subscription does happen on the thread from computation scheduler, as we can read in the output. But once we emit values — even from a new thread — RxJava will happily take those emissions and pass them further in the stream. It won’t change the thread. Because why would it? The subscribeOn operator did its work — the subscription already happened on a different thread. There's no reason RxJava would change the thread again.

The example above might’ve been obvious, because we’ve clearly created a new thread. A straightforward conclusion is to simply not do that, we rely on RxJava to do the threading after all! But what about third party libraries? A good example is Google Play Services that we can use on Android to get user’s location. Let’s have a look at how we’d wrap callback-based location emissions in an Observable:

Pretty standard — we create a callback instance, set cancellable to unregister it when necessary, start listening to updates and push them to the stream. When we receive new location, we want to fetch some data for it from the network, so we're using map operator for that. Finally, we want everything to happen on an io thread, so we subscribe on it, and should be fine, right?

But what is this Looper.getMainLooper() ? Seems like the location client uses the main thread somewhere, no big deal, eh? Only as you probably expect, it is pretty big deal if it affects the thread on which the emissions happen.

If we have a look at the location client’s documentation, we can read:

Callbacks for LocationCallback will be made on the specified thread, which must already be a prepared looper thread.

Which means, our onLocationResult method is called on the main thread. Thus no matter what thread we subscribe on, the emissions always happen on the main thread, because that's where the library executes the callback. The result is we’re emitting location on the main thread, and so the network call is also happening on the main thread. Fortunately, the above code would crash at that point, but sometimes this behavior is subtly blocking UI or causing difficult to debug bugs.

To achieve the behavior we want, we have to change the thread after the emission, using observeOn operator. This case is especially tricky if you want to encapsulate your code and always emit values on the subscription scheduler. Because an observable doesn't have explicit access to the subscription scheduler, it's difficult to emit values on it. There's a somewhat related issue that's unfortunately closed for some time now.

PublishSubject

There’s another case when the thread change and subscribeOn behavior isn't obvious: subjects. With subjects we first establish a stream, to which we can later emit values. It's not difficult to figure out what can go wrong here: if we call onNext on the subject from a certain thread, that thread is used for the emission, regardless of the scheduler we used to subscribe. This is also stated right there in the documentation.

Let’s consider an example:

The result aren’t surprising by now:

[Test worker] 1 - I want this happen on an IO thread

[Test worker] 1 - I want this happen on a new thread

[Test worker] 2 - I want this happen on an IO thread

[Test worker] 2 - I want this happen on a new thread

[Test worker] 3 - I want this happen on an IO thread

[Test worker] 3 - I want this happen on a new thread

To achieve what we want, which is handling the emissions on a separate thread, we need to use observeOn in place of the subscribeOn calls. This makes RxJava change the thread after the item was emitted, regardless of the emission thread:

The result then is as expected:

[RxNewThreadScheduler-1] 1 - I want this happen on a new thread

[RxCachedThreadScheduler-1] 1 - I want this happen on an IO thread

[RxNewThreadScheduler-1] 2 - I want this happen on a new thread

[RxCachedThreadScheduler-1] 2 - I want this happen on an IO thread

[RxNewThreadScheduler-1] 3 - I want this happen on a new thread

[RxCachedThreadScheduler-1] 3 - I want this happen on an IO thread

BehaviorSubject

Things are slightly bit different with BehaviorSubject s, where subscribeOn can sometimes affect the emission thread. Let's recall how a BehaviorSubject works: for each onNext called on the subject, it will notify all current subscribers. At the same time it will cache this last emission, and replay it for the new subscribers. So, there are two scenarios in which an emission can happen: when we explicitly call onNext , and when someone subscribes to the subject.

Let’s see this on an example:

and the output:

[Test worker] 1 - First observer

[RxCachedThreadScheduler-1] 1 - Second observer

[RxNewThreadScheduler-2] 1 - Third observer

[Test worker] 2 - First observer

[Test worker] 2 - Second observer

[Test worker] 2 - Third observer

[Test worker] 3 - First observer

[Test worker] 3 - Second observer

[Test worker] 3 - Third observer

Now that’s interesting! 1 was emitted on 3 different threads! Why? Let's go one by one:

First observer was already subscribed when we emitted first value. Since the subscription code has already completed by the time we called onNext , there's no reason for subscribe scheduler to have any effect. When we call onNext in this scenario, it's analogous to how PublishSubject works — observers are notified on the same thread that called onNext .

Second and third observers only subscribed after the initial onNext . This is where BehaviorSubject 's behavior kicks in — for any new subscriber, it will replay the latest emission. So for these two observers it's no longer our onNext call that emits 1 . Rather, the BehaviorSubject that notices it has some value cached, and emits it as part of the subscription. As such, the subscription scheduler is respected, and the observers are notified on a thread that it provides.

All subsequent emissions happen outside of subscription, so the values are emitted on the same thread as onNext , again, similarly to how PublishSubject works. Also the same rules apply — specifiying scheduler using observeOn will ensure you receive the value on the appropriate thread.

Conclusion

I hope this clears up some of RxJava’s behavior and helps avoid subtle threading bugs. While they won’t necessarily crash your app, they may silently crowd your main thread or cause unexpected results. When using streams spanning across multiple app layers, subjects and third party libraries, it’s quite easy to miss that the thread isn’t necessarily what you expect it to be. So keep an eye out, don’t change the thread yourself (unless you know exactly what you’re doing), and pay attention to third party libraries’ threading behavior!