Rxandroid What's the difference between SubscribeOn and ObserveOn
SubscribeOn specify the Scheduler on which an Observable will operate. ObserveOn specify the Scheduler on which an observer will observe this Observable.
So basically SubscribeOn is mostly subscribed (executed) on a background thread ( you do not want to block the UI thread while waiting for the observable) and also in ObserveOn you want to observe the result on a main thread...
If you are familiar with AsyncTask then SubscribeOn is similar to doInBackground method and ObserveOn to onPostExecute...
In case you find the above answer full of jargons:
tl;dr
Observable.just("Some string")
.map(str -> str.length())
.observeOn(Schedulers.computation())
.map(length -> 2 * length)
.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())
.subscribe(---)
Observe an observable, perform the map function in an IO thread (since we are
"subscribingOn"
that thread), now switch to a Computation Thread and performmap(length -> 2 * length)
function now make sure you Observe the output on Main thread but perform all the tasks defined undersubscribe()
in a IO thread.
Aanyway,
observeOn()
simply changes the thread of all operators further Downstream. People usually have this misconception that observeOn
also acts as upstream, but it doesn't.
The below example will explain it better..
Observable.just("Some string") // UI
.map(str -> str.length()) // UI
.observeOn(Schedulers.computation()) // Changing the thread
.map(length -> 2 * length) // Computation
.subscribe(---)
subscribeOn()
only influences the thread which is going to be used when Observable is going to get subscribed to and it will stay on it downstream.
Observable.just("Some String") // Computation
.map(str -> str.length()) // Computation
.map(length -> 2 * length) // Computation
.subscribeOn(Schedulers.computation()) // -- changing the thread
.subscribe(number -> Log.d("", "Number " + number));// Computation
Position does not matter (
subscribeOn()
)
Why? Because it affects only the time of subscription.
Methods that obey the contact with
subscribeOn
-> Basic example : Observable.create
All the work specified inside the create
body will run on the thread specified in subscribeOn
.
Another example: Observable.just
,Observable.from
or Observable.range
Note: All those methods accept values, so do not use blocking methods to create those values, as subscribeOn won't affect it.
If you want to use blocking functions, use
Observable.defer(() -> Obervable.just(blockingMenthod())));
Important Fact:
subscribeOn does not work with Subjects
Multiple
subscribeOn
:
If there are multiple instances of subscribeOn
in the stream, only the first one has a practical effect.
Subscribe &
subscribeOn
People think that subscribeOn
has something to do with Observable.subscribe
, but it doesn't have anything special to do with it.
It only affects the subscription phase.
Source : Tomek Polański (Medium)
Summary
- Use
observeOn
to set threads for callbacks "further down the stream (below it)", such as code blocks insidedoOnNext
ormap
. - Use
subscribeOn
to set threads for initializations "upstream (above it)", such asdoOnSubscribe
,Observable.just
orObservable.create
. - Both methods can be called multiple times, with each call overwriting previous ones. Position matters.
Let's walk through this topic with an example: we want to find the length of the string "user1032613". This is not an easy task for computers, so it's only natural that we perform the intense calculation in a background thread, to avoid freezing the app.
observeOn
We can call observeOn
as many times as we like, and it controls which thread all callbacks below it will run. It's easy to use, and works just as you'd expect.
For example, we will show a progress bar on the main UI thread, then do intensive/blocking operations in another thread, then come back to the main UI thread to update the result:
Observable.just("user1032613")
.observeOn(mainThread) // set thread for operation 1
.doOnNext {
/* operation 1 */
print("display progress bar")
progressBar.visibility = View.VISIBLE
}
.observeOn(backThread) // set thread for operation 2 and 3
.map {
/* operation 2 */
print("calculating")
Thread.sleep(5000)
it.length
}
.doOnNext {
/* operation 3 */
print("finished calculating")
}
.observeOn(mainThread) // set thread for operation 4
.doOnNext {
/* operation 4 */
print("hide progress bar and display result")
progressBar.visibility = View.GONE
resultTextView.text = "There're $it characters!"
}
.subscribe()
In the above example, /* operation 1 */
is ran in the mainThread
because we set it using observeOn(mainThread)
on the line right above it; then we switch to backThread
by calling observeOn
again, so /* operation 2 */
will run there. Because we didn't change it before chaining /* operation 3 */
, it will run in the back thread as well, just like /* operation 2 */
; finally we call observeOn(mainThread)
again, to make sure /* operation 4 */
updates the UI from the main thread.
subscribeOn
So we've learned observeOn
sets threads for subsequent callbacks. What else are we missing? Well, the Observable
itself, and its methods such as just()
, create()
, subscribe()
and so on, are also code that needs to be executed. This is how objects are passed along the stream. We use subscribeOn
to set threads for code related to Observable
itself.
If we remove all the callbacks (controlled by observeOn
discussed earlier), we are left with the "skeleton code" that will, by default, run on whichever thread the code is written in (probably main thread):
Observable.just("user1032613")
.observeOn(mainThread)
.doOnNext {
}
.observeOn(backThread)
.map {
}
.doOnNext {
}
.observeOn(mainThread)
.doOnNext {
}
.subscribe()
If we aren't happy about this empty skeleton code running on main thread, we can use subscribeOn
to change it. For example, maybe the first line Observable.just("user1032613")
isn't as simple as creating a stream from my user name - maybe it's a string from the Internet, or perhaps you are using doOnSubscribe
for some other intensive operations. In that case, you can call subscribeOn(backThread)
to put some of the code in another thread.
Where to put subscribeOn
At the time of writing this answer, there are some misconceptions saying "only call it once", "position does not matter", and "if you call it multiple times, only the first time counts". After lots of researches and experiments, it turns out subscribeOn
can be usefully called multiple times.
Because Observable
uses Builder Pattern (fancy name for "chaining methods one after another"), subscribeOn
is applied in reverse order. Therefore, this method sets the thread for code above it, exactly the opposite of observeOn
.
We can experiment this using doOnSubscribe
method. This method is triggered on the subscription event, and it runs on the thread set by subscribeOn
:
Observable.just("user1032613")
.doOnSubscribe {
print("#3 running on main thread")
}
.subscribeOn(mainThread) // set thread for #3 and just()
.doOnNext {
}
.map {
}
.doOnSubscribe {
print("#2 running on back thread")
}
.doOnNext {
}
.subscribeOn(backThread) // set thread for #2 above
.doOnNext {
}
.doOnSubscribe {
print("#1 running on default thread")
}
.subscribe()
It might be easier to follow the logic, if you read the above example from bottom to top, just like how Builder Pattern executes the code.
In this example, the first line Observable.just("user1032613")
is run in the same thread as print("#3")
because there are no more subscribeOn
in-between them. This creates the illusion of "only the first call matters" for people who only care about code inside just()
or create()
. This quickly falls apart once you start doing more.
Footnote:
Threads and print()
functions in the examples are defined, for brevity, as follows:
val mainThread = AndroidSchedulers.mainThread()
val backThread = Schedulers.computation()
private fun print(msg: String) = Log.i("", "${Thread.currentThread().name}: $msg")