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