Flowable

在以前的RxJava版本中，只有一个基类可以处理可感知背压的源和不可感知背压的源Observable。RxJava 2在这两种源之间引入了明显的区别-背压感知源现在使用专用类Flowable表示。Observable不支持背压，因此，我们应该将其用于仅消耗而无法影响的资源。

创建与控制

// Simple Flowable
Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3, 4);

// Flowable from Observable
Observable<Integer> integerObservable = Observable.just(1, 2, 3);
Flowable<Integer> integerFlowable = integerObservable
  .toFlowable(BackpressureStrategy.BUFFER);

// from FlowableOnSubscribe
FlowableOnSubscribe<Integer> flowableOnSubscribe
 = flowable -> flowable.onNext(1);
Flowable<Integer> integerFlowable = Flowable
  .create(flowableOnSubscribe, BackpressureStrategy.BUFFER);

BackpressureStrategy

诸如toFlowable()或create()之类的某些方法采用BackpressureStrategy作为参数。BackpressureStrategy是一个枚举，它定义了我们将应用于Flowable的背压行为。它可以缓存或丢弃事件，或者根本不执行任何行为，在最后一种情况下，我们将负责使用反压运算符定义它。BackpressureStrategy与RxJava先前版本中存在的BackpressureMode相似。

RxJava 2提供了五种不同的策略。

Buffer

如果我们使用BackpressureStrategy.BUFFER，则源将缓冲所有事件，直到订阅者可以使用它们：

public void thenAllValuesAreBufferedAndReceived() {
    List testList = IntStream.range(0, 100000)
      .boxed()
      .collect(Collectors.toList());

    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.BUFFER)
      .observeOn(Schedulers.computation()).test();

    testSubscriber.awaitTerminalEvent();

    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertEquals(testList, receivedInts);
}

这类似于在Flowable上调用onBackpressureBuffer()方法，但是它不允许显式定义缓冲区大小或onOverflow操作。

Drop

我们可以使用BackpressureStrategy.DROP丢弃无法消耗的事件，而不是对其进行缓冲。同样，这类似于在Flowable上使用onBackpressureDrop()：

public void whenDropStrategyUsed_thenOnBackpressureDropped() {

    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.DROP)
      .observeOn(Schedulers.computation())
      .test();
    testSubscriber.awaitTerminalEvent();
    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertThat(receivedInts.size() < testList.size());
    assertThat(!receivedInts.contains(100000));
 }

Latest

使用BackpressureStrategy.LATEST将强制源仅保留最新事件，从而在使用者无法跟上时覆盖任何先前的值：

public void whenLatestStrategyUsed_thenTheLastElementReceived() {

    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.LATEST)
      .observeOn(Schedulers.computation())
      .test();

    testSubscriber.awaitTerminalEvent();
    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertThat(receivedInts.size() < testList.size());
    assertThat(receivedInts.contains(100000));
 }

当我们查看代码时，BackpressureStrategy.LATEST和BackpressureStrategy.DROP看起来非常相似。但是，BackpressureStrategy.LATEST将覆盖我们的Subscriber无法处理的元素，并仅保留最新的元素。另一方面，BackpressureStrategy.DROP将丢弃无法处理的元素，这意味着不一定会发出最新元素。

Error

当我们使用BackpressureStrategy.ERROR时，我们只是在说我们不希望发生背压。因此，如果使用者跟不上源，就应该抛出MissingBackpressureException：

public void whenErrorStrategyUsed_thenExceptionIsThrown() {
    Observable observable = Observable.range(1, 100000);
    TestSubscriber subscriber = observable
      .toFlowable(BackpressureStrategy.ERROR)
      .observeOn(Schedulers.computation())
      .test();

    subscriber.awaitTerminalEvent();
    subscriber.assertError(MissingBackpressureException.class);
}

Missing

如果使用BackpressureStrategy.MISSING，则源将推送元素而不会丢弃或缓冲。在这种情况下，下游将必须处理溢出：

public void whenMissingStrategyUsed_thenException() {
    Observable observable = Observable.range(1, 100000);
    TestSubscriber subscriber = observable
      .toFlowable(BackpressureStrategy.MISSING)
      .observeOn(Schedulers.computation())
      .test();
    subscriber.awaitTerminalEvent();
    subscriber.assertError(MissingBackpressureException.class);
}

在我们的测试中，我们同时针对ERROR和MISSING策略都缺少MissingbackpressureException。因为当源的内部缓冲区溢出时，它们都将引发此类异常。但是，值得注意的是，两者都有不同的目的。当我们完全不期望背压时，应该使用前一种，并且我们希望源抛出异常以防万一。如果我们不想在创建Flowable时指定默认行为，则可以使用后一种方法。

最近更新于0001-01-01