RxJava2外传Ⅰ：官方README翻译

发表于 | 分类于 | 阅读次数:

有些翻译仍不准确，会持续改进。

术语

上游，下游

RxJava中的数据流包括一个数据源、0个或多个中间步骤、一个数据消费者或组合子步骤(其中的步骤负责以某种方式使用数据流):

1
2
3
source.operator1().operator2().operator3().subscribe(consumer);

source.flatMap(value -> source.operator1().operator2().operator3());

如果我们想象自己站在operator2上，向左看向source，叫做上游；向右看向subscriber/consumer，叫做下游。当像下面这样每个元素单独写一行时看的更加明显：

1
2
3
4
5
source
  .operator1()
  .operator2()
  .operator3()
  .subscribe(consumer)

运动的对象

在RxJava的文档中， emission, emits, item, event, signal, data and message都是近义词，都表示沿着数据流移动的对象。

背压

当数据流通过异步步骤运行时，每个步骤可能以不同的速度执行不同的事情。为了避免那些由于临时缓冲或需要跳过/删除数据而导致内存使用量增加的步骤被淹没，因此应用了所谓的背压，这是一种流控制形式，从而步骤可以表示准备处理多少项数据。使用背压允许当前步骤在通常无法知道上游将发送多少项数据的情况下限制数据流的内存使用。

在RxJava中，Flowable类被设计成支持背压，Observable类专用于非背压操作。Single, Maybe and Completable也不支持背压。

装配时间Assembly time

通过应用各式各样的中间操作符来准备数据流，发生在装配时间。

1
2
3
4
Flowable<Integer> flow = Flowable.range(1, 5)
.map(v -> v * v)
.filter(v -> v % 3 == 0)
;

在当前点上，数据还没有流动，也没有发生副作用。

订阅时间 Subscription time

这是在内部建立处理步骤链的流上调用subscribe()时的临时状态:

1
flow.subscribe(System.out::println)

这时会触发订阅副作用。有些源在这种状态下会立即阻塞或开始发送数据项。

运行时 Runtime

这是数据流主动发出数据项、错误或完成信号时的状态:

1
2
3
4
5
6
7
8
9
10
11
Observable.create(emitter -> {
     while (!emitter.isDisposed()) {
         long time = System.currentTimeMillis();
         emitter.onNext(time);
         if (time % 2 != 0) {
             emitter.onError(new IllegalStateException("Odd millisecond!"));
             break;
         }
     }
})
.subscribe(System.out::println, Throwable::printStackTrace);

实际上，这是上面给定示例的主体执行的时候。

简单的后台计算

RxJava的一个常见用例是在后台线程上运行一些计算、网络请求，并在UI线程上显示结果(或错误)

1
2
3
4
5
6
7
8
9
10
11
import io.reactivex.schedulers.Schedulers;

Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
})
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.single())
  .subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000); // <--- wait for the flow to finish

这种链式调用方法的形式称为流式API，类似于builder模式。然而，RxJava的响应类型是不可变的;每个方法调用都返回一个添加了行为的新的Flowable。我们可以把上面的例子改写为下面这样：

1
2
3
4
5
6
7
8
9
10
11
12
Flowable<String> source = Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
});

Flowable<String> runBackground = source.subscribeOn(Schedulers.io());

Flowable<String> showForeground = runBackground.observeOn(Schedulers.single());

showForeground.subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000);

通常，您可以通过subscribeOn将计算或阻塞IO移动到其他线程。一旦数据准备好，您就可以通过observeOn确保它们在前台或GUI线程上得到处理。

调度者Schedulers

RxJava操作符不直接使用Thread或ExecutorServices，而是使用所谓的调度器Scheduler，这些调度器将并发源抽象到统一API后面。RxJava 2提供了几个可通过scheduler类访问的标准调度器。

  • Schedulers.computation(): 在后台的固定数量的专用线程上运行计算密集型工作。大多数异步操作符将此作为其默认调度程序。
  • Schedulers.io(): 在一组动态变化的线程上运行类似I/ o或阻塞操作。
  • Schedulers.single(): 以顺序和FIFO方式在单个线程上运行工作。
  • Schedulers.trampoline(): 在参与的线程中以顺序和FIFO方式运行工作，通常用于测试目的。

这些在所有JVM平台上都可用，但是在某些特定的平台上，例如android，有它们特有的调度器：AndroidSchedulers.mainThread(), SwingScheduler.instance() or JavaFXSchedulers.gui().

此外，还可以通过Scheduler .from(Executor)将现有的Executor(及其子类型，如ExecutorService)包装到调度器Schedulers中。例如，可以使用它来拥有更大但仍然固定的线程池(不同于分别使用compute()和io())。

上面例子结尾处的Thread.sleep(2000);是有意为之。在RxJava中，默认调度程序在守护线程上运行，这意味着一旦Java主线程退出，它们就会全部停止，后台计算可能永远不会发生。在这个示例场景中，休眠一段时间可以让您看到控制台上数据流流的输出。

流中的并发

RxJava中的流本质上是顺序的，它们被划分为可以彼此并发运行的处理阶段

1
2
3
4
Flowable.range(1, 10)
  .observeOn(Schedulers.computation())
  .map(v -> v * v)
  .blockingSubscribe(System.out::println);

这个示例流将computation调度器上的数字从1平方到10，并在主线程(更准确地说，是blockingSubscribe的调用线程)上处理结果。然而lambda表达式v -> v * v并不是并行运行的。它在同一个计算线程上依次接收1到10的值。

并行处理

并行处理数字1到10稍微复杂一些:

1
2
3
4
5
6
7
Flowable.range(1, 10)
  .flatMap(v ->
      Flowable.just(v)
        .subscribeOn(Schedulers.computation())
        .map(w -> w * w)
  )
  .blockingSubscribe(System.out::println);

实际上，RxJava中的并行性意味着运行独立的流并将它们的结果合并回单个流。操作符flatMap首先将1到10的每个数字映射到它自己的Flowable中，运行它们并合并计算的平方。

但是，请注意，flatMap不保证任何顺序，来自内部流的最终结果可能是交错的。有其他的操作符可供选择：

  • concatMap 它每次映射并运行一个内部流
  • concatMapEager 它“同时”运行所有内部流，但是输出流将按照这些内部流创建的顺序输出。

或者，Flowable.parallel()操作符和ParallelFlowable类型可以帮助实现相同的并行处理模式:

1
2
3
4
5
6
Flowable.range(1, 10)
  .parallel()
  .runOn(Schedulers.computation())
  .map(v -> v * v)
  .sequential()
  .blockingSubscribe(System.out::println);

依赖子流

flatMap是一个强大的操作符，在很多情况下都有帮助。例如，给定一个返回Flowable的服务，我们希望使用第一个服务发出的值调用另一个服务:

1
2
3
4
5
6
7
8
Flowable<Inventory> inventorySource = warehouse.getInventoryAsync();

inventorySource.flatMap(inventoryItem ->
    erp.getDemandAsync(inventoryItem.getId())
    .map(demand 
        -> System.out.println("Item " + inventoryItem.getName() + " has demand " + demand));
  )
  .subscribe();

延续Continuations

有时，当一个项变得可用时，人们希望依赖它执行一些计算。这有时称为延续，根据将要发生的情况和涉及的类型，可能需要不同的操作符来实现。

依赖

最典型的场景是给定一个值，调用另一个服务，等待并使用其结果继续:

1
2
3
service.apiCall()
.flatMap(value -> service.anotherApiCall(value))
.flatMap(next -> service.finalCall(next))

通常情况下，后面的序列也需要来自前面映射的值。这可以通过将外部flatMap 移动到上一个flatMap的内部来实现，例如:

1
2
3
4
5
service.apiCall()
.flatMap(value ->
    service.anotherApiCall(value)
    .flatMap(next -> service.finalCallBoth(value, next))
)

在这里，由于lambda变量捕获，原始value将在内部flatMap中可用。

非依赖

在其他场景中，第一个源/数据流的结果是不相关的，我们希望使用独立的另一个源继续。flatMap 也可以胜任：

1
2
3
Observable continued = sourceObservable.flatMapSingle(ignored -> someSingleSource)
continued.map(v -> v.toString())
  .subscribe(System.out::println, Throwable::printStackTrace);

然而，在这种情况下，延续保持Observable ，而不是可能更合适的Single.(这是可以理解的，因为从flatMapSingle的角度来看，sourceObservable是一个多值源，因此映射也可能导致多个值)

虽然通常有一种方法更有表现力(也更低的开销)，即使用Completable作为中介，然后使用它的操作符andThen来继续:

1
2
3
4
sourceObservable
  .ignoreElements()           // returns Completable
  .andThen(someSingleSource)
  .map(v -> v.toString())

sourceObservable和someSingleSource之间唯一的依赖关系是前者应该正常完成，以便使用后者。

延迟依赖

有时，前一个序列和新序列之间存在隐式的数据依赖关系，由于某些原因，该依赖关系没有通过“常规通道”流动。有人会倾向于这样写延续:

1
2
3
4
5
6
7
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.just(count.get()))
  .subscribe(System.out::println);

不幸的是，这输出0，因为Single.just(count.get())是在数据流尚未运行的assembly time期间计算的。我们需要一些东西，来推迟Single来源的计算，直到主要来源完成的运行时:

1
2
3
4
5
6
7
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.defer(() -> Single.just(count.get())))
  .subscribe(System.out::println);

或者

1
2
3
4
5
6
7
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.fromCallable(() -> count.get()))
  .subscribe(System.out::println);

类型转换

有时，源或服务返回的类型与应该使用它的流不同。 例如，在上面的inventory例子中，getDemandAsync 可能返回一个Single<DemandRecord>. 如果代码示例保持不变，则会导致编译时错误(但是，通常会出现关于缺少重载的误导性错误消息)。

在这种情况下，通常有两种方式来修复转换: 1) 转换为所需类型 2) 查找并使用支持不同类型的特定操作符的重载。

转换为所需类型

每个响应基类都具有可以执行此类转换(包括协议转换)的操作符，以匹配其他类型。 下面的矩阵显示了可用的转换选项:

Flowable Observable Single Maybe Completable
Flowable toObservable first, firstOrError, single, singleOrError, last, lastOrError1 firstElement, singleElement, lastElement ignoreElements
Observable toFlowable2 first, firstOrError, single, singleOrError, last, lastOrError1 firstElement, singleElement, lastElement ignoreElements
Single toFlowable3 toObservable toMaybe ignoreElement
Maybe toFlowable3 toObservable toSingle ignoreElement
Completable toFlowable toObservable toSingle toMaybe

1: 当将一个多值源转换为一个单值源时，你应该决定使用多个源中的哪一个值作为结果。

2: 把一个Observable变成Flowable 需要一个额外的决定: 如何处理Observable源中潜在的无约束流? 通过BackpressureStrategy参数或标准的Flowable操作符(如onBackpressureBuffer、onBackpressureDrop、onbackpressurerelatest,这些操作符还允许进一步定制反压力行为。)可以使用几种策略(如缓冲、删除、保持最新状态)来处理。

3: 当只有(最多)一个源数据项时，背压没有问题，因为它可以一直存储到下游准备使用为止。

使用具有所需类型的重载

许多常用的操作符都有可以处理其他类型的重载。它们通常以目标类型的后缀命名:

Operator Overloads
flatMap flatMapSingle, flatMapMaybe, flatMapCompletable, flatMapIterable
concatMap concatMapSingle, concatMapMaybe, concatMapCompletable, concatMapIterable
switchMap switchMapSingle, switchMapMaybe, switchMapCompletable

这些操作符具有后缀而不是简单地使用具有不同签名的相同名称的原因是类型擦除。 Java认为像operator(Function<T, Single<R>>) 和 operator(Function<T, Maybe<R>>)这样的签名是相同的(与c#不同)，并且由于擦除的原因，这两operator最终会成为具有相同签名的重复方法。

操作符命名约定

命名在编程中是最困难的事情之一，因为命名要求不长、具有表现力、容易捕捉和容易记忆。 不幸的是，目标语言(和已经存在的约定)在这方面可能不会提供太多的帮助(不可用的关键字、类型擦除、类型歧义等等)。

不可用关键字

在原始的Rx.NET中，发出单个项然后完成的运算符称为 Return(T).由于Java约定是以小写字母开头的方法名，因此它应该是return(T)，但这是Java中的关键字，因此不可用。 因此，RxJava选择将这个操作符命名为 just(T). 操作符Switch也存在同样的限制，必须将其命名为 switchOnNext.另一个例子是Catch，它被命名为 onErrorResumeNext.

类型消除

许多期望用户提供返回响应类型的函数的操作符无法重载，因为函数Function<T, X>周围的类型擦除将这些方法签名转换为重复。RxJava通过添加类型后缀来命名这些操作符:

1
2
3
Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper)

Flowable<R> flatMapMaybe(Function<? super T, ? extends MaybeSource<? extends R>> mapper)

类型歧义

尽管某些操作符在类型擦除方面没有问题，但是是在使用Java 8和lambdas的情况下，它们的签名可能会变得含糊不清。例如，concatWith以各种其他反应性基类型作为参数(为了在底层实现中提供方便和性能优势)，会出现一些重载:

1
2
3
Flowable<T> concatWith(Publisher<? extends T> other);

Flowable<T> concatWith(SingleSource<? extends T> other);

Publisher和SingleSource都以函数接口的形式出现(带有一个抽象方法的类型)，并可能鼓励用户尝试提供lambda表达式:

1
2
someSource.concatWith(s -> Single.just(2))
.subscribe(System.out::println, Throwable::printStackTrace);

不幸的是，这种方法不起作用，示例根本不打印2。 实际上，从2.1.10版本开始，它甚至不能编译，因为至少有4个concatWith重载存在，编译器发现上面的代码不明确。

在这种情况下，用户可能希望延迟一些计算，直到someSource完成，因此正确的明确操作符应该是defer:

1
2
someSource.concatWith(Single.defer(() -> Single.just(2)))
.subscribe(System.out::println, Throwable::printStackTrace);

有时，添加后缀是为了避免逻辑歧义，这些歧义可能会编译，但会在流中产生错误的类型:

1
2
3
Flowable<T> merge(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> mergeArray(Publisher<? extends T>... sources);

当函数接口类型涉及到类型参数T时，这也会变得含糊不清。

错误处理

数据流可能失败，此时错误被发送到使用者。 但是，有时多个源可能会失败，这时可以选择是否等待所有源完成或失败。为了表示这种机会，许多操作符的名称都添加了DelayError单词的后缀(而其他操作符的重载中则添加了delayError或delayErrors boolean标志):

1
2
3
Flowable<T> concat(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources);

当然，各种各样的后缀可能出现在一起:

1
Flowable<T> concatArrayEagerDelayError(Publisher<? extends T>... sources);

基类vs基类型

由于基类上的静态方法和实例方法的数量太多，因此可以认为基类很重。 RxJava 2的设计深受响应流规范的影响，因此，该库为每种响应类型提供了一个类和一个接口:

Type Class Interface Consumer
0..N backpressured Flowable Publisher1 Subscriber
0..N unbounded Observable ObservableSource2 Observer
1 element or error Single SingleSource SingleObserver
0..1 element or error Maybe MaybeSource MaybeObserver
0 element or error Completable CompletableSource CompletableObserver

1The org.reactivestreams.Publisher是外部响应流库的一部分。它是通过受响应流规范控制的标准化机制与其他响应库进行交互的主要类型。

2接口的命名约定是将Source追加到半传统的类名中。 因为Publisher是由响应流库提供的，所以没有FlowableSource(而且子类型对互操作也没有帮助)。 然而，这些接口并不是响应流规范意义上的标准接口，而且目前仅针对RxJava。

坚持原创技术分享，您的支持将鼓励我继续创作！