一. ParallelFlowable
1.1 ParallelFlowable介紹
RxJava 2.0.5 版本新增了 ParallelFlowable API艇棕，它允許并行地執(zhí)行一些運(yùn)算符，譬如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中很多典型的運(yùn)算符(如take, skip等等)是不可用的又沾。
在RxJava中沒(méi)有ParallelObservable弊仪，因?yàn)樵赗xJava 2之后Observable不再支持背壓。然而在并行處理中背壓是必不可少的杖刷，否則會(huì)淹沒(méi)在并行運(yùn)算符的內(nèi)部隊(duì)列中励饵。正是因?yàn)閿?shù)據(jù)的處理在一個(gè)線程上比較緩慢，所以我們才會(huì)使用并行滑燃。
同理役听，也沒(méi)有ParallelSingle、ParallelCompletable以及ParallelMaybe表窘。
1.2 ParallelFlowable實(shí)現(xiàn)并行
類似Java 8的并行流典予，在相應(yīng)的操作符上調(diào)用Flowable的parallel()就會(huì)返回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()調(diào)用了ParallelFlowable.from(@NonNull Publisher<? extends T> source)
public final ParallelFlowable<T> parallel() {
return ParallelFlowable.from(this);
}
ParallelFlowable的from()方法瘤袖，通過(guò)Publisher并以循環(huán)的方式在多個(gè)“軌道”（CPU數(shù)）上消費(fèi)它。
/**
* 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());
}
在默認(rèn)情況下昂验，并行級(jí)別被設(shè)置為可用CPU的數(shù)量(Runtime.getRuntime().availableProcessors())捂敌，并且順序源的預(yù)取量設(shè)置為Flowable.bufferSize()。 兩者都可以通過(guò)重載parallel()來(lái)指定既琴。
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);
}
在最后占婉，如果已經(jīng)使用了必要的并行操作，您可以通過(guò)ParallelFlowable.sequential()操作符返回到順序流甫恩。
parallelFlowable
......
.sequential()
.subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
}
});
1.3 ParallelFlowable 與 Scheduler
ParallelFlowable遵循與Flowable相同的異步原理逆济，因此parallel()本身不引入順序源的異步消耗，只準(zhǔn)備并行流磺箕。但是可以通過(guò)runOn(Scheduler)操作符定義異步纹腌。這一點(diǎn)跟Flowable很大不同，F(xiàn)lowable是使用subscribeOn滞磺、observeOn操作符升薯。
ParallelFlowable<Integer> psource = source.runOn(Schedulers.io());
runOn()可以指定prefetch的數(shù)量。
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來(lái)實(shí)現(xiàn)并行。Flowable.flatMap實(shí)現(xiàn)并行和Observable.flatMap實(shí)現(xiàn)并行的原理是一樣的蛛枚。
那何時(shí)使用flatMap進(jìn)行并行處理比較好谅海，何時(shí)使用ParallelFlowable比較好呢？
RxJava 本質(zhì)上是連續(xù)的蹦浦，借助flatMap操作符進(jìn)行分離和加入一個(gè)序列可能會(huì)變得復(fù)雜扭吁，并引起一定的開(kāi)銷。 但是如果使用ParallelFlowable的話開(kāi)銷會(huì)更小盲镶。
ParallelFlowable具有有限的操作符侥袜。 因此，如果你有一些特殊的操作需要并行執(zhí)行溉贿，而這些操作不能用ParallelFlowable所支持的操作符表達(dá)枫吧，那么肯定應(yīng)該使用基于Flowable.flatMap來(lái)實(shí)現(xiàn)并行。
