-
Notifications
You must be signed in to change notification settings - Fork 7.6k
Parallel flows
Version 2.0.5 introduced the ParallelFlowable
API that allows parallel execution of a few select operators such as map
, filter
, concatMap
, flatMap
, collect
, reduce
and so on. Note that is a parallel mode for Flowable
(a sub-domain specific language) instead of a new reactive base type.
Consequently, several typical operators such as take
, skip
and many others are not available and there is no ParallelObservable
because backpressure is essential in not flooding the internal queues of the parallel operators as by expectation, we want to go parallel because the processing of the data is slow on one thread.
The easiest way of entering the parallel world is by using Flowable.parallel
:
ParallelFlowable<Integer> source = Flowable.range(1, 1000).parallel();
By default, the parallelism level is set to the number of available CPUs (Runtime.getRuntime().availableProcessors()
) and the prefetch amount from the sequential source is set to Flowable.bufferSize()
(128). Both can be specified via overloads of parallel()
.
ParallelFlowable
follows the same principles of parametric asynchrony as Flowable
does, therefore, parallel()
on itself doesn't introduce the asynchronous consumption of the sequential source but only prepares the parallel flow; the asynchrony is defined via the runOn(Scheduler)
operator.
ParallelFlowable<Integer> psource = source.runOn(Schedulers.io());
The parallelism level (ParallelFlowable.parallelism()
) doesn't have to match the parallelism level of the Scheduler
. The runOn
operator will use as many Scheduler.Worker
instances as defined by the parallelized source. This allows ParallelFlowable
to work for CPU intensive tasks via Schedulers.computation()
, blocking/IO bound tasks through Schedulers.io()
and unit testing via TestScheduler
. You can specify the prefetch amount on runOn
as well.
Once the necessary parallel operations have been applied, you can return to the sequential Flowable
via the ParallelFlowable.sequential()
operator.
Flowable<Integer> result = psource.filter(v -> v % 3 == 0).map(v -> v * v).sequential();
Note that sequential
doesn't guarantee any ordering between values flowing through the parallel operators.
TBD
Copyright (c) 2016-present, RxJava Contributors.
Twitter @RxJava | Gitter @RxJava