RxJava 之 ParallelFlowable
RxJava 之 ParallelFlowable
parallelflowable.sequential.png
一. ParallelFlowable
1.1 ParallelFlowable介绍
RxJava 2.0.5 版本新增了 ParallelFlowable API，它允许并行地执行一些运算符，譬如map、filter、concatMap、flatMap、collect、reduce等。
/**
* Abstract base class for Parallel publishers that take an array of Subscribers.
* <p>
* Use {@code from()} to start processing a regular Publisher in 'rails'.
* Use {@code runOn()} to introduce where each 'rail' should run on thread-vise.
* Use {@code sequential()} to merge the sources back into a single Flowable.
*
* <p>History: 2.0.5 - experimental
* @param <T> the value type
* @since 2.1 - beta
*/
@Beta
public abstract class ParallelFlowable<T> {
...
}ParallelFlowable是并行的Flowable版本，并不是新增的被观察者类型。在ParallelFlowable中很多典型的运算符(如take, skip等等)是不可用的。
在RxJava中没有ParallelObservable，因为在RxJava 2之后Observable不再支持背压。然而在并行处理中背压是必不可少的，否则会淹没在并行运算符的内部队列中。正是因为数据的处理在一个线程上比较缓慢，所以我们才会使用并行。
同理，也没有ParallelSingle、ParallelCompletable以及ParallelMaybe。
1.2 ParallelFlowable实现并行
类似Java 8的并行流，在相应的操作符上调用Flowable的parallel()就会返回ParallelFlowable。
ParallelFlowable parallelFlowable = Flowable.range(1,100).parallel();
parallelFlowable
.runOn(Schedulers.io())
.map(new Function<Integer, Object>() {
@Override
public Object apply(@NonNull Integer integer) throws Exception {
return integer.toString();
}
})
.sequential()
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String str) throws Exception {
System.out.println(str);
}
});其中，parallel()调用了ParallelFlowable.from(@NonNull Publisher<? extends T> source)
public final ParallelFlowable<T> parallel() {
return ParallelFlowable.from(this);
}ParallelFlowable的from()方法，通过Publisher并以循环的方式在多个“轨道”（CPU数）上消费它。
/**
* Take a Publisher and prepare to consume it on multiple 'rails' (number of CPUs)
* in a round-robin fashion.
* @param <T> the value type
* @param source the source Publisher
* @return the ParallelFlowable instance
*/
@CheckReturnValue
public static <T> ParallelFlowable<T> from(@NonNull Publisher<? extends T> source) {
return from(source, Runtime.getRuntime().availableProcessors(), Flowable.bufferSize());
}在默认情况下，并行级别被设置为可用CPU的数量(Runtime.getRuntime().availableProcessors())，并且顺序源的预取量设置为Flowable.bufferSize()。 两者都可以通过重载parallel()来指定。
public final ParallelFlowable<T> parallel(int parallelism) {
ObjectHelper.verifyPositive(parallelism, "parallelism");
return ParallelFlowable.from(this, parallelism);
}
public final ParallelFlowable<T> parallel(int parallelism, int prefetch) {
ObjectHelper.verifyPositive(parallelism, "parallelism");
ObjectHelper.verifyPositive(prefetch, "prefetch");
return ParallelFlowable.from(this, parallelism, prefetch);
}在最后，如果已经使用了必要的并行操作，您可以通过ParallelFlowable.sequential()操作符返回到顺序流。
parallelFlowable
......
.sequential()
.subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
}
});1.3 ParallelFlowable 与 Scheduler
ParallelFlowable遵循与Flowable相同的异步原理，因此parallel()本身不引入顺序源的异步消耗，只准备并行流。但是可以通过runOn(Scheduler)操作符定义异步。这一点跟Flowable很大不同，Flowable是使用subscribeOn、observeOn操作符。
ParallelFlowable<Integer> psource = source.runOn(Schedulers.io());runOn()可以指定prefetch的数量。
public final ParallelFlowable<T> runOn(@NonNull Scheduler scheduler) {
return runOn(scheduler, Flowable.bufferSize());
}
public final ParallelFlowable<T> runOn(@NonNull Scheduler scheduler, int prefetch) {
ObjectHelper.requireNonNull(scheduler, "scheduler");
ObjectHelper.verifyPositive(prefetch, "prefetch");
return RxJavaPlugins.onAssembly(new ParallelRunOn<T>(this, scheduler, prefetch));
}二. ParallelFlowable的操作符
并非所有的顺序操作在并行世界中都是有意义的。
目前ParallelFlowable只支持如下的操作：
map,
filter,
flatMap,
concatMap,
reduce,
collect,
sort,
toSortedList,
compose,
doOnCancel, doOnError, doOnComplete, doOnNext, doOnSubscribe, doAfterTerminate, doOnRequest
这些ParallelFlowable可用的操作符，使用方法跟Flowable中的用法是一样的。
三. ParallelFlowable 和 Flowable.flatMap 比较
上一篇文章RxJava 并行操作介绍了使用Observable.flatMap来实现并行。Flowable.flatMap实现并行和Observable.flatMap实现并行的原理是一样的。
那何时使用flatMap进行并行处理比较好，何时使用ParallelFlowable比较好呢？
RxJava 本质上是连续的，借助flatMap操作符进行分离和加入一个序列可能会变得复杂，并引起一定的开销。 但是如果使用ParallelFlowable的话开销会更小。
ParallelFlowable具有有限的操作符。 因此，如果你有一些特殊的操作需要并行执行，而这些操作不能用ParallelFlowable所支持的操作符表达，那么肯定应该使用基于Flowable.flatMap来实现并行。