Reactive Programming on Android with RxJava

Adding Asynchronicity
There are many ways we can add asynchronicity so that the function within
.create()
is invoked
on a separate thread from the calling thread. As mentioned in Chapter 2, RxJava is agnostic as to
how asynchronicity is accomplished. One way is to make use of a plain old
Thread
. Let’s take our
previous example… instead of immediately calling
.onNext()
in the function provided in
.create()
,
we will…
1. … wrap the
.onNext()
and
.onComplete()
calls in a
Runnable
,
2. … provide the
Runnable
to a new
Thread
,
3. … and then start the Thread.
1
Observable
<
Integer
>
integerObservable
=
Observable
.
create
(
source
-> {
2
Log
.
d
(
TAG
,
"In subscribe"
);
3
new
Thread
(() -> {
4
source
.
onNext
(1);
5
source
.
onNext
(2);
6
source
.
onNext
(3);
7
source
.
onComplete
();
8
}).
start
();
9
});
10
Log
.
d
(
TAG
,
"Created Observable"
);
11
12
Log
.
d
(
TAG
,
"Subscribing to Observable"
);
13
integerObservable
.
subscribe
(
i
->
Log
.
d
(
TAG
,
"In onNext(): "
+
i
));
14
15
Log
.
d
(
TAG
,
"Finished"
);

---

Chapter 4: Multithreading
49
With this change, we now get:
1
Created Observable
2
Subscribing to Observable
3
In subscribe
4
Finished
5
In
onNext
(): 1
6
In
onNext
(): 2
7
In
onNext
(): 3
As we can see, the “Finished” statement is invoked before the “onNext()” statements. In fact, the
order of these statements are no longer deterministic and our original log sequence is still possible;
it all depends on when the OS decides to perform a context switch, which can change from one
execution to the next.
This approach to non-blocking subscription is perfectly valid as long as an invocation to
.onNext()
,
.onError()
, or
.onComplete()
is performed from a single thread and that it is thread-safe. If this
contract is breached, some
Operator
s that depend on this assumption may fail to function (i.e. we
cannot emit events from multiple threads).
The above example was used to show that a plain old
Thread
(and any other Java/Android threading
construct) can be used to add asynchronicity to RxJava. However, just because using these is
possible
does not mean that they
should
be used. The preferred way of achieving concurrency in RxJava is
actually through the use of
Scheduler
s.
Schedulers
A
Scheduler
is a multithreading construct introduced in RxJava that can run a scheduled unit of
work on a thread. Simplistically, you can think of a
Scheduler
as a thread pool, and when a task
needs to be executed, it takes a single thread from its pool and runs the necessary task.
Scheduler
s are
used in conjunction with
.subscribeOn()
and
.observeOn()
, two
Operator
s that specify where a