A multithreaded app has two or more parts that can run in parallel. This lets the app make better use of the cores inside the device CPU. This lets it get tasks done faster and leads to a smoother and more responsive experience for the user.
Coding for concurrency in Java can be painful, but thanks to RxJava, it is now much easier to do. With RxJava, you just need to declare the thread on which you want the task to be executed (declaratively) instead of creating and managing threads (imperatively).
RxJava makes use of Schedulers along with the subscribeOn() and observeOn() concurrency operators to achieve this. In this tutorial, you’ll learn about Schedulers, the subscribeOn() operator, the observeOn() operator, and also how to leverage the flatMap() operator to achieve concurrency. But first, let’s start with Schedulers in RxJava.
Prerequisites
To follow along with this tutorial, you should be familiar with:
- RxJava 2 on Android
- lambda expressions
Check out our other posts to get up to speed on the basics of RxJava and lambda expressions.
Android SDKGet Started With RxJava 2 for Android
Android SDKCoding Functional Android Apps in Kotlin: Lambdas, Null Safety & More
Schedulers in RxJava 2
Schedulers in RxJava are used to execute a unit of work on a thread. A Scheduler provides an abstraction to the Android and Java threading mechanism. When you want to run a task and you make use of a Scheduler to execute that task, the Scheduler goes to its thread pool (a collection of threads that are ready for use) and then runs the task in an available thread.
You can also specify that a task should run in one specific thread. (There are two operators, subscribeOn() and observeOn(), which can be used to specify on which thread from the Scheduler thread pool the task should be executed.)
As you know, in Android, long-running processes or CPU-intensive tasks should not be run on the main thread. If a subscription by an Observer to an Observable is conducted on the main thread, any associated operator will run on the main thread also. In the case of a long-running task (e.g. performing a network request) or a CPU-intensive task (e.g. image transformation), this will block the UI until the task is finished, leading to the awful ANR (Application Not Responding) dialog and the app crashing. These operators can instead be switched over to another thread with the observeOn() operator.
In the next section, we are going to explore the different kinds of Schedulers and their uses.
Types of Schedulers
Here are some of the types of Schedulers available in RxJava and RxAndroid to indicate the kind of thread to execute tasks on.
-
Schedulers.immediate(): returns aSchedulerwhich executes the work instantly in the current thread. Be aware that this will block the current thread, so it should be used with caution. -
Schedulers.trampoline(): schedules tasks in the current thread. These tasks are not executed immediately but instead are executed after the thread has finished its current tasks. This is different fromSchedulers.immediate()because instead of executing a task immediately, it waits for the current tasks to complete. -
Schedulers.newThread(): fires up a new thread and returns aSchedulerto execute the task in the new thread for eachObserver. You should be careful using this because the new thread is not reused afterwards but instead is destroyed. -
Schedulers.computation(): this gives us aSchedulerthat is intended for computationally intensive work such as image transformation, complex calculations, etc. This operation fully employs the CPU cores. ThisScheduleruses a fixed thread pool size which is dependent on the CPU cores for optimal usage. You should be careful not to create more threads than the available CPU cores because this can reduce performance. -
Schedulers.io(): creates and returns aSchedulerdesignated for I/O-bound work such as performing asynchronous network calls or reading and writing to the database. These tasks are not CPU-intensive or else make use ofSchedulers.computation(). -
Schedulers.single(): creates and returns aSchedulerand executes several tasks sequentially in a single thread. -
Schedulers.from(Executor executor): this will create aSchedulerthat will execute a task or unit of work on the givenExecutor. -
AndroidSchedulers.mainThread(): this will create aSchedulerthat executes the task on the Android application main thread. This scheduler type is provided by theRxAndroidlibrary.
The subscribeOn() Operator
By using the subscribeOn() concurrency operator, you specify that the Scheduler should perform the operation in the Observable upstream. It will then push the values to the Observers using the same thread. Now let’s see a practical example:
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
public class MainActivity extends AppCompatActivity {
private static final String[] STATES = { "Lagos", "Abuja", "Abia",
"Edo", "Enugu", "Niger", "Anambra"};
private Disposable mDisposable = null;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Observable observable = Observable.create(dataSource())
.subscribeOn(Schedulers.newThread())
.doOnComplete(() -> Log.d("MainActivity", "Complete"));
mDisposable = observable.subscribe(s -> {
Log.d("MainActivity", "received " + s + " on thread " + Thread.currentThread().getName());
});
}
private ObservableOnSubscribe dataSource() {
return(emitter -> {
for(String state : STATES) {
emitter.onNext(state);
Log.d("MainActivity", "emitting " + state + " on thread " + Thread.currentThread().getName());
Thread.sleep(600);
}
emitter.onComplete();
});
}
@Override
protected void onDestroy() {
if (mDisposable != null && !mDisposable.isDisposed()) {
mDisposable.dispose();
}
super.onDestroy();
}
}
In the code above, we have a static ArrayList which contains some states in Nigeria. We also have a field which is of type Disposable. We get the Disposable instance by calling Observable.subscribe(), and we’ll use it later when we call the dispose() method to release any resources that were used. This helps to prevent memory leaks. Our dataSource() method (which can return data from a remote or local database source) will return ObservableOnSubscribe: this is required for us to create our own Observable later using the method Observable.create().
Inside the dataSource() method, we loop through the array, emitting each element to the Observers by calling emitter.onNext(). After each value is emitted, we sleep the thread so as to simulate intensive work being performed. Finally, we call the onComplete() method to signal to the Observers that we are done passing values and they shouldn’t expect any more.
Now, our dataSource() method should not be executed on the main UI thread. But how is this specified? In the example above, we provided Schedulers.newThread() as an argument to subscribeOn(). This means that the dataSource() operation will be run in a new thread. Note also that in the example above, we have just one Observer. If we had multiple Observers, each of them would get its own thread.
So that we can see this working, our Observer prints out the values it gets in its onNext() method from the Observable.
When we run this and view our logcat on Android Studio, you can see that the emissions from the dataSource() method to the Observer happened on the same thread—RxNewThreadScheduler-1—in which the Observer received them.
If you don’t specify the .subscribeOn() method after the Observable.create() method, it will be executed on the current thread—which in our case is the main thread, thereby blocking the app UI.
There are some important details you should be aware of concerning the subscribeOn() operator. You should only have one subscribeOn() in the Observable chain; adding another one anywhere in the chain will have no effect at all. The recommended place to put this operator is as close to the source as possible for the sake of clarity. In other words, place it first in the operator chain.
Observable.create(dataSource())
.subscribeOn(Schedulers.computation()) // this has effect
.subscribeOn(Schedulers.io()) // has no effect
.doOnNext(s -> {
saveToCache(s); // executed on Schedulers.computation()
})
The observeOn() Operator
As we saw, the subscribeOn() concurrency operator will instruct the Observable which Scheduler to use to push emissions forward along the Observable chain to the Observers.
The job of the observeOn() concurrency operator, on the other hand, is to switch the subsequent emissions over to another thread or Scheduler. We use this operator to control on what thread downstream consumers will receive the emissions. Let’s see a practical example.
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.TextView;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
public class ObserveOnActivity extends AppCompatActivity {
private Disposable mDisposable = null;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
TextView textView = (TextView) findViewById(R.id.tv_main);
Observable observable = Observable.create(dataSource())
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.doOnComplete(() -> Log.d("ObserveOnActivity", "Complete"));
mDisposable = observable.subscribe(s -> {
Log.d("ObserveOnActivity", "received " + s + " on thread " + Thread.currentThread().getName());
textView.setText(s);
});
}
private ObservableOnSubscribe dataSource() {
return(emitter -> {
Thread.sleep(800);
emitter.onNext("Value");
Log.d("ObserveOnActivity", "dataSource() on thread " + Thread.currentThread().getName());
emitter.onComplete();
});
}
// ...
}
In the code above, we used the observeOn() operator and then passed the AndroidSchedulers.mainThread() to it. What we have done is to switch the thread from Schedulers.newThread() to the Android main thread. This is necessary because we want to update the TextView widget, and can only do that from the main UI thread. Note that if you don’t switch over to the main thread when you try to update the TextView widget, the app will crash and throw a CalledFromWrongThreadException.
Unlike the subscribeOn() operator, the observeOn() operator can be applied multiple times in the operator chain, thereby changing the Scheduler more than once.
Observableobservable = Observable.create(dataSource()) .subscribeOn(Schedulers.newThread()) .observeOn(Schedulers.io()) .doOnNext(s -> { saveToCache(s); Log.d("ObserveOnActivity", "doOnNext() on thread " + Thread.currentThread().getName()); }) .observeOn(AndroidSchedulers.mainThread()) .doOnComplete(() -> Log.d("ObserveOnActivity", "Complete"));
This code has two observeOn() operators. The first one uses the Schedulers.io(), which means that the saveToCache() method will be executed on the Schedulers.io() thread. After that, it then switches over to the AndroidSchedulers.mainThread() where Observers will receive the emissions from the upstream.
Concurrency With the flatMap() Operator
The flatMap() operator is another very powerful and important operator that can be used to achieve concurrency. The definition according to the official documentation is as follows:
Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable.
Let’s take a look at a practical example that uses this operator:
//...
@Override
protected void onCreate(Bundle savedInstanceState) {
// ...
final String[] states = {"Lagos", "Abuja", "Imo", "Enugu"};
Observable statesObservable = Observable.fromArray(states);
statesObservable.flatMap(
s -> Observable.create(getPopulation(s))
).subscribe(pair -> Log.d("MainActivity", pair.first + " population is " + pair.second));
}
private ObservableOnSubscribe getPopulation(String state) {
return(emitter -> {
Random r = new Random();
Log.d("MainActivity", "getPopulation() for " + state + " called on " + Thread.currentThread().getName());
emitter.onNext(new Pair(state, r.nextInt(300000 - 10000) + 10000));
emitter.onComplete();
});
}
}
This will print the following on Android Studio logcat:
getPopulation() for Lagos called on main Lagos population is 80362 getPopulation() for Abuja called on main Abuja population is 132559 getPopulation() for Imo called on main Imo population is 34106 getPopulation() for Enugu called on main Enugu population is 220301
From the result above, you can see that the results we got were in the same order as in the array. Also, the getPopulation() method for each state was processed on the same thread—the main thread. This makes the output result slow because they were processed sequentially on the main thread.
Now, in order for us to achieve concurrency with this operator, we want the getPopulation() method for each state (emissions from the statesObservable) to be processed on different threads. Doing this will lead to faster processing. We’ll use the flatMap() operator to do this because it creates a new Observable for each emission. We then apply the subscribeOn() concurrency operator to each one, passing a Scheduler to it.
statesObservable.flatMap(
s -> Observable.create(getPopulation(s))
.subscribeOn(Schedulers.io())
).subscribe(pair -> Log.d("MainActivity", pair.first + " population is " + pair.second));
As each emission produces an Observable, the flatMap() operator’s job is to merge them together and then send them out as a single stream.
getPopulation() for Lagos called on RxCachedThreadScheduler-1 Lagos population is 143965 getPopulation() for Abuja called on RxCachedThreadScheduler-2 getPopulation() for Enugu called on RxCachedThreadScheduler-4 Abuja population is 158363 Enugu population is 271420 getPopulation() for Imo called on RxCachedThreadScheduler-3 Imo population is 81564
In the result above, we can observe that each state’s getPopulation() method was processed on different threads. This makes the processing much faster, but also observe that the emissions from the flatMap() operator which were received by the Observer are not in the same order as the original emissions upstream.
Conclusion
In this tutorial, you learned about handling concurrency using RxJava 2: what it is, the different Schedulers available, and how to use the subscribeOn() and observeOn() concurrency operators. I also showed you how to use the flatMap() operator to achieve concurrency.
In the meantime, check out some of our other courses and tutorials on the Java language and Android app development!
Android SDKIntroduction to Android Architecture Components
Android SDKWhat Are Android Instant Apps?
iOS SDK3 Terrible Mistakes of iOS Developers-
Android SDKCoding Functional Android Apps in Kotlin: Lambdas, Null Safety & More
-
Android SDKGet Started With RxJava 2 for Android
Powered by WPeMatico
