RxJava mistakes: (not so) parallel execution of blocking tasks
In this short post, I want to write you about some stupid mistake I made when I wanted to execute two blocking tasks in parallel using RxJava. I spent about half an hour debugging my code trying to figure out why it’s executed synchronously despite the fact that I use subscribeOn method. Now if I think about it it was pretty obvious but I think most bugs seems obvious after you solve them 😉.
My task was pretty simple. I had two long long running blocks of code which I put into separate functions called processData1() and processData2(). None of the functions return any value. I wanted to execute them on separate background threads in parallel and notify when they both complete or at least one of it fails.
The first step would be to wrap the functions into Completables using Completable.fromAction() method. Completable emits either an exception or completes when the task is finished so it‘s perfect for wrapping functions which don’t return any value.
Merging it together
To execute both Completables in parallel they have to be merged using (not surprisingly) merge operator.
Merge combines multiple Observables (or in that case Completables) into one by merging their emissions like shown on the diagram:
For two completables the diagram can be simplified since completable doesn’t emit any value. In that case, merge function creates a new completable instance which completes when both completables completes or emits an error.
Specify the scheduler
To execute the tasks on a separate thread it’s convenient to use subscribeOn function which allows specifying Scheduler on which a Completable will operate. I passed Schedulers.io() scheduler so both actions will be executed on the thread from the IO pool.
Resulting code (with a bug 🐛)
The resulting code is really simple and looks like this:
As I mentioned earlier that code has a subtle bug in it. If you run that code with some prints you’ll see that both tasks are executed on the thread from IO pool but they’re both executed on the same thread.
The reason is really simple: mergeWith function combines two completables into one and returns a new Completable instance which subscribes to both completables. The scheduler won’t be applied to each of the completables but only to completable returned by merge.
Corrected version
To both tasks work in parallel subscribeOn should be called on each of the completables that would be merged.
And the output finally is correct, both tasks run in parallel on the separate thread from the io pool.
What about other operators?
The same rule applies not only to the merge operator but also to zip operator which you can use to run a code from two Singles or Observables in parallel and combine their results.
Code for parallel execution of two zipped Singles which emit some Strings would look like this:
