一. 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来实现并行。
