大佬们，一波RxJava 3.0来袭，请做好准备~

Rouse

发布于 2019-07-17 17:48:01

1.9K00

代码可运行

文章被收录于专栏：Android补给站Android补给站

运行总次数：0

代码可运行

GitCode8

读完需要

分钟

速读仅需5分钟

作者：GitCode8 链接：https://juejin.im/post/5d1eeffe6fb9a07f0870b4e8

前言

每个Android开发者，都是爱RxJava的,简洁线程切换和多网络请求合并，再配合Retrofit,简直是APP开发的福音。不知不觉，RxJava一路走来，已经更新到第三大版本了。不像RxJava 2对RxJava 1那么残忍，RxJava 3对RxJava 2的兼容性还是挺好的，目前并没有做出很大的更改。RxJava2到2020年12月31号不再提供支持，错误时同时在2.x和3.x修复，但新功能只会在3.x上添加。

同时，希望通过本文，能知道垃圾箱颜色分类。

作为尝鲜，赶紧品尝吧。

主要变化

2.1

主要特点

单一依赖：Reactive-Streams
继续支持Java 6+和Android 2.3+
修复了API错误和RxJava 2的许多限制
旨在替代RxJava 2，具有相对较少的二进制不兼容更改
提供Java 8 lambda友好的API
关于并发源（线程，池，事件循环，光纤，演员等）的不同意见
异步或同步执行
参数化并发的虚拟时间和调度程序
为测试schedulers，consumers和plugin hooks提供测试和诊断支持

2.2

与RxJava的主要区别是：

将eagerTruncate添加到replay运算符，以便head节点将在截断时丢失
它保留的项引用
新增 X.fromSupplier()
使用 Scheduler 添加 concatMap，保证 mapper 函数的运行位置
新增 startWithItem 和 startWithIterable
ConnectableFlowable/ConnetableFlowable 重新设计
将 as() 并入 to()
更改 Maybe.defaultIfEmpty() 以返回 Single
用 Supplier 代替 Callable
将一些实验操作符推广到标准
从某些主题/处理器中删除 getValues()
删除 replay(Scheduler) 及其重载
删除 dematerialize()
删除 startWith(T|Iterable)
删除 as()
删除 Maybe.toSingle(T)
删除 Flowable.subscribe(4 args)
删除 Observable.subscribe(4 args)
删除 Single.toCompletable()
删除 Completable.blockingGet()

到这里就结束了，想知道的都知道了。

入门

3.1

添加依赖

1implementation "io.reactivex.rxjava3:rxjava:3.0.0-RC0"

不好意思哦，还没看到RxAndroid出3.0，这就很尴尬了...

3.2

一些概念

上流、下流

在RxJava,数据以流的方式组织。也就是说，Rxjava包括一个源的数据流，数据流后跟着消费者的零个到多个消费数据流步骤。

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

在上文代码中，对于operator2来说，在它前面叫做上流，在它后面的叫做下流。憋住，别笑，真的是下流来的。

流的对象

在RxJava的文档中，emission, emits, item, event, signal, data and message都被认为在数据流中被传递的数据对象。

背压(Backpressure)

当数据流通过异步的步骤执行时，这些步骤的执行速度可能不一致。也就是说上流数据发送太快，下流没有足够的能力去处理。为了避免这种情况，一般要么缓存上流的数据，要么抛弃数据。但这种处理方式，有时会带来很大的问题。为此，RxJava带来了backpressure的概念。背压是一种流量的控制步骤，在不知道上流还有多少数据的情形下控制内存的使用，表示它们还能处理多少数据。

支持背压的有Flowable类，不支持背压的有Observable，Single, Maybe and Completable类。

线程调度器（Schedulers）

对于我们Android开发来说，最喜欢的就是它简洁切换线程的操作。RxJava通过调度器来方便线程的切换。

Schedulers.computation(): 适合运行在密集计算的操作，大多数异步操作符使用该调度器。
Schedulers.io():适合运行I/0和阻塞操作.
Schedulers.single():适合需要单一线程的操作
Schedulers.trampoline(): 适合需要顺序运行的操作

在不同平台还有不同的调度器，例如Android的主线程：AndroidSchedulers.mainThread()

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

在 RxJava 3 可以发现有以下几个基类（跟RxJava 2是一致的吧）：

io.reactivex.Flowable：发送0个N个的数据，支持Reactive-Streams和背压
io.reactivex.Observable：发送0个N个的数据，不支持背压，
io.reactivex.Single：只能发送单个数据或者一个错误
io.reactivex.Completable：没有发送任何数据，但只处理 onComplete 和 onError 事件。
io.reactivex.Maybe：能够发射0或者1个数据，要么成功，要么失败。

不建议再往下看了，建议点赞或收藏...下文关于操作符内容太多了等需要了，再来查阅下班时间还是好好护发吧

操作符

4.1

实用操作符

ObserveOn

指定观察者的线程，例如在Android访问网络后，数据需要主线程消费，那么将观察者的线程切换到主线就需要ObserveOn操作符。每次指定一次都会生效。

subscribeOn

指定被观察者的线程，即数据源发生的线程。例如在Android访问网络时，需要将线程切换到子线程。多次指定只有第一次有效。

doOnEach

数据源（Observable）每发送一次数据，就调用一次。

doOnNext

数据源每次调用onNext() 之前都会先回调该方法。

doOnError

数据源每次调用onError() 之前会回调该方法。

doOnComplete

数据源每次调用onComplete() 之前会回调该方法

doOnSubscribe

数据源每次调用onSubscribe() 之后会回调该方法

doOnDispose

数据源每次调用dispose() 之后会回调该方法

其他的见官网吧，不难

实用操作符:

https://github.com/ReactiveX/RxJava/wiki/Observable-Utility-Operators

4.2

对数据源过滤操作符

主要讲对数据源进行选择和过滤的常用操作符

skip（跳过）

可以作用于Flowable,Observable，表示源发射数据前，跳过多少个。例如下面跳过前四个：

 1Observable<Integer> source = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 2
 3source.skip(4)
 4    .subscribe(System.out::print);
 5
 6打印结果：5678910
 7
 8Observable<Integer> source = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 9
10source.skipLast(4)
11    .subscribe(System.out::print);
12
13打印结果：1 2 3 4 5 6

skipLast(n)操作表示从流的尾部跳过n个元素。

debounce（去抖动）

可作用于Flowable,Observable。在Android开发，通常为了防止用户重复点击而设置标记位，而通过RxJava的debounce操作符可以有效达到该效果。在规定时间内，用户重复点击只有最后一次有效，

 1Observable<String> source = Observable.create(emitter -> {
 2    emitter.onNext("A");
 3
 4    Thread.sleep(1_500);
 5    emitter.onNext("B");
 6
 7    Thread.sleep(500);
 8    emitter.onNext("C");
 9
10    Thread.sleep(250);
11    emitter.onNext("D");
12
13    Thread.sleep(2_000);
14    emitter.onNext("E");
15    emitter.onComplete();
16});
17
18source.subscribeOn(Schedulers.io())
19        .debounce(1, TimeUnit.SECONDS)
20        .blockingSubscribe(
21                item -> System.out.print(item+" "),
22                Throwable::printStackTrace,
23                () -> System.out.println("onComplete"));
24
25打印：A D E onComplete

上文代码中，数据源以一定的时间间隔发送A,B,C,D,E。操作符debounce的时间设为1秒，发送A后1.5秒并没有发射其他数据，所以A能成功发射。发射B后，在1秒之内，又发射了C和D,在D之后的2秒才发射E,所有B、C都失效，只有D有效；而E之后已经没有其他数据流了，所有E有效。

distinct（去重）

可作用于Flowable,Observable，去掉数据源重复的数据。

1Observable.just(2, 3, 4, 4, 2, 1)
2        .distinct()
3        .subscribe(System.out::println);
4
5// 打印:2 3 4 2 1
6Observable.just(1, 1, 2, 1, 2, 3, 3, 4)
7        .distinctUntilChanged()
8        .subscribe(System.out::print);
9//打印：1 2 1 2 3 4

distinctUntilChanged()去掉相邻重复数据。

elementAt（获取指定位置元素）

可作用于Flowable,Observable，从数据源获取指定位置的元素，从0开始。

 1 Observable.just(2,4,3,1,5,8)
 2        .elementAt(0)
 3        .subscribe(integer -> 
 4         Log.d("TAG","elmentAt->"+integer));
 5打印：2
 6
 7Observable<String> source = Observable.just("Kirk", "Spock", "Chekov", "Sulu");
 8Single<String> element = source.elementAtOrError(4);
 9
10element.subscribe(
11    name -> System.out.println("onSuccess will not be printed!"),
12    error -> System.out.println("onError: " + error));
13打印：onSuccess will not be printed!

elementAtOrError：指定元素的位置超过数据长度，则发射异常。

filter（过滤）

可作用于 Flowable,Observable,Maybe,Single。在filter中返回表示发射该元素，返回false表示过滤该数据。

1Observable.just(1, 2, 3, 4, 5, 6)
2        .filter(x -> x % 2 == 0)
3        .subscribe(System.out::print);
4打印：2 4 6

first(第一个)

作用于 Flowable,Observable。发射数据源第一个数据，如果没有则发送默认值。

 1Observable<String> source = Observable.just("A", "B", "C");
 2Single<String> firstOrDefault = source.first("D");
 3firstOrDefault.subscribe(System.out::println);
 4打印：A
 5
 6Observable<String> emptySource = Observable.empty();
 7Single<String> firstOrError = emptySource.firstOrError();
 8firstOrError.subscribe(
 9        element -> System.out.println("onSuccess will not be printed!"),
10        error -> System.out.println("onError: " + error));
11打印：onError: java.util.NoSuchElementException

和firstElement的区别是first返回的是Single，而firstElement返回Maybe。firstOrError在没有数据会返回异常。

last(最后一个)

last、lastElement、lastOrError与fist、firstElement、firstOrError相对应。

 1Observable<String> source = Observable.just("A", "B", "C");
 2Single<String> lastOrDefault = source.last("D");
 3lastOrDefault.subscribe(System.out::println);
 4//打印:C
 5
 6Observable<String> source = Observable.just("A", "B", "C");
 7Maybe<String> last = source.lastElement();
 8last.subscribe(System.out::println);
 9//打印:C
10
11Observable<String> emptySource = Observable.empty();
12Single<String> lastOrError = emptySource.lastOrError();
13lastOrError.subscribe(
14        element -> System.out.println("onSuccess will not be printed!"),
15        error -> System.out.println("onError: " + error));
16// 打印：onError: java.util.NoSuchElementException

ignoreElements 作用于Flowable、Observable。ignoreElement作用于Maybe、Single。两者都是忽略掉数据，返回完成或者错误时间。

 1Single<Long> source = Single.timer(1, TimeUnit.SECONDS);
 2Completable completable = source.ignoreElement();
 3completable.doOnComplete(() -> System.out.println("Done!"))
 4        .blockingAwait();
 5// 1秒后打印：Donde!
 6
 7Observable<Long> source = Observable.intervalRange(1, 5, 1, 1, TimeUnit.SECONDS);
 8Completable completable = source.ignoreElements();
 9completable.doOnComplete(() -> System.out.println("Done!"))
10        .blockingAwait();
11// 五秒后打印：Done!

ofType（过滤掉类型）

作用于Flowable、Observable、Maybe、过滤掉类型。

1Observable<Number> numbers = Observable.just(1, 4.0, 3, 2.71, 2f, 7);
2Observable<Integer> integers = numbers.ofType(Integer.class);
3integers.subscribe((Integer x) -> System.out.print(x+" "));
4//打印:1 3 7

sample

作用于Flowable、Observable，在一个周期内发射最新的数据。

 1Observable<String> source = Observable.create(emitter -> {
 2    emitter.onNext("A");
 3
 4    Thread.sleep(500);
 5    emitter.onNext("B");
 6
 7    Thread.sleep(200);
 8    emitter.onNext("C");
 9
10    Thread.sleep(800);
11    emitter.onNext("D");
12
13    Thread.sleep(600);
14    emitter.onNext("E");
15    emitter.onComplete();
16});
17
18source.subscribeOn(Schedulers.io())
19        .sample(1, TimeUnit.SECONDS)
20        .blockingSubscribe(
21                item -> System.out.print(item+" "),
22                Throwable::printStackTrace,
23                () -> System.out.print("onComplete"));
24
25// 打印：C D onComplete

与debounce的区别是，sample是以时间为周期的发射，一秒又一秒内的最新数据。而debounce是最后一个有效数据开始。

throttleFirst & throttleLast & throttleWithTimeout

作用于Flowable、Observable。throttleLast与smaple一致，而throttleFirst是指定周期内第一个数据。throttleWithTimeout与debounce一致。

 1Observable<String> source = Observable.create(emitter -> {
 2    emitter.onNext("A");
 3
 4    Thread.sleep(500);
 5    emitter.onNext("B");
 6
 7    Thread.sleep(200);
 8    emitter.onNext("C");
 9
10    Thread.sleep(800);
11    emitter.onNext("D");
12
13    Thread.sleep(600);
14    emitter.onNext("E");
15    emitter.onComplete();
16});
17
18source.subscribeOn(Schedulers.io())
19        .throttleFirst(1, TimeUnit.SECONDS)
20        .blockingSubscribe(
21                item -> System.out.print(item+" "),
22                Throwable::printStackTrace,
23                () -> System.out.print(" onComplete"));
24//打印:A D onComplete
25
26source.subscribeOn(Schedulers.io())
27        .throttleLast(1, TimeUnit.SECONDS)
28        .blockingSubscribe(
29                item -> System.out.print(item+" "),
30                Throwable::printStackTrace,
31                () -> System.out.print(" onComplete"));
32
33// 打印:C D onComplete

throttleLatest

之所以拿出来单独说，我看不懂官网的解释。然后看别人的文章：throttleFirst+throttleLast的组合？开玩笑的吧。个人理解是：如果源的第一个数据总会被发射，然后开始周期计时，此时的效果就会跟throttleLast一致。

 1Observable<String> source = Observable.create(emitter -> {
 2            emitter.onNext("A");
 3
 4            Thread.sleep(500);
 5            emitter.onNext("B");
 6
 7            Thread.sleep(200);
 8            emitter.onNext("C");
 9
10            Thread.sleep(200);
11            emitter.onNext("D");
12
13            Thread.sleep(400);
14            emitter.onNext("E");
15
16            Thread.sleep(400);
17            emitter.onNext("F");
18
19            Thread.sleep(400);
20            emitter.onNext("G");
21
22            Thread.sleep(2000);
23            emitter.onComplete();
24        });
25        source.subscribeOn(Schedulers.io())
26        .throttleLatest(1, TimeUnit.SECONDS)
27        .blockingSubscribe(
28            item -> Log.e("RxJava",item),
29                 Throwable::printStackTrace,
30            () -> Log.e("RxJava","finished"));

打印结果：

take & takeLast

作用于Flowable、Observable，take发射前n个元素;takeLast发射后n个元素。

1Observable<Integer> source = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
2
3source.take(4)
4    .subscribe(System.out::print);
5//打印:1 2 3 4
6
7source.takeLast(4)
8    .subscribe(System.out::println);
9//打印:7 8 9 10

timeout（超时）

作用于Flowable、Observable、Maybe、Single、Completabl。后一个数据发射未在前一个元素发射后规定时间内发射则返回超时异常。

 1Observable<String> source = Observable.create(emitter -> {
 2    emitter.onNext("A");
 3
 4    Thread.sleep(800);
 5    emitter.onNext("B");
 6
 7    Thread.sleep(400);
 8    emitter.onNext("C");
 9
10    Thread.sleep(1200);
11    emitter.onNext("D");
12    emitter.onComplete();
13});
14
15source.timeout(1, TimeUnit.SECONDS)
16        .subscribe(
17                item -> System.out.println("onNext: " + item),
18                error -> System.out.println("onError: " + error),
19                () -> System.out.println("onComplete will not be printed!"));
20// 打印:
21// onNext: A
22// onNext: B
23// onNext: C
24// onError: java.util.concurrent.TimeoutException: 
25            The source did not signal an event for 1 seconds 
26            and has been terminated

4.3

连接操作符

通过连接操作符，将多个被观察数据（数据源）连接在一起。

startWith

可作用于Flowable、Observable。将指定数据源合并在另外数据源的开头。

 1Observable<String> names = Observable.just("Spock", "McCoy");
 2Observable<String> otherNames = Observable.just("Git", "Code","8");
 3names.startWith(otherNames).subscribe(item -> Log.d(TAG,item));
 4
 5//打印：
 6RxJava: Git
 7RxJava: Code
 8RxJava: 8
 9RxJava: Spock
10RxJava: McCo

merge

可作用所有数据源类型，用于合并多个数据源到一个数据源。

 1Observable<String> names = Observable.just("Hello", "world");
 2Observable<String> otherNames = Observable.just("Git", "Code","8");
 3
 4Observable.merge(names,otherNames).subscribe(name -> Log.d(TAG,name));
 5
 6//也可以是
 7//names.mergeWith(otherNames).subscribe(name -> Log.d(TAG,name));
 8
 9//打印：
10RxJava: Hello
11RxJava: world
12RxJava: Git
13RxJava: Code
14RxJava: 8

merge在合并数据源时，如果一个合并发生异常后会立即调用观察者的onError方法，并停止合并。可通过mergeDelayError操作符，将发生的异常留到最后处理。

 1Observable<String> names = Observable.just("Hello", "world"); 
 2Observable<String> otherNames = Observable.just("Git", "Code","8");
 3Observable<String> error = Observable.error(    
 4                            new NullPointerException("Error!"));
 5Observable.mergeDelayError(names,error,otherNames).subscribe(
 6    name -> Log.d(TAG,name), e->Log.d(TAG,e.getMessage()));
 7
 8//打印：
 9RxJava: Hello
10RxJava: world
11RxJava: Git
12RxJava: Code
13RxJava: 8
14RxJava: Error!

zip

可作用于Flowable、Observable、Maybe、Single。将多个数据源的数据一个一个的合并在一起哇。当其中一个数据源发射完事件之后，若其他数据源还有数据未发射完毕，也会停止。

1Observable<String> names = Observable.just("Hello", "world");
2Observable<String> otherNames = Observable.just("Git", "Code", "8");
3names.zipWith(otherNames, (first, last) -> first + "-" + last)
4       .subscribe(item -> Log.d(TAG, item));
5
6//打印：
7RxJava: Hello-Git
8RxJava: world-Code

combineLatest

可作用于Flowable, Observable。在结合不同数据源时，发射速度快的数据源最新item与较慢的相结合。如下时间线，Observable-1发射速率快，发射了65，Observable-2才发射了C, 那么两者结合就是C5。

switchOnNext

一个发射多个小数据源的数据源，这些小数据源发射数据的时间发生重复时，取最新的数据源。

4.4

变换操作符

变化数据源的数据，并转化为新的数据源。

buffer

作用于Flowable、Observable。指将数据源拆解含有长度为n的list的多个数据源，不够n的成为一个数据源。

1Observable.range(0, 10)
2    .buffer(4)
3    .subscribe((List<Integer> buffer) -> System.out.println(buffer));
4
5// 打印:
6// [0, 1, 2, 3]
7// [4, 5, 6, 7]
8// [8, 9]

cast

作用于Flowable、Observable、Maybe、Single。将数据元素转型成其他类型,转型失败会抛出异常。

 1Observable<Number> numbers = Observable.just(1, 4.0, 3f, 7, 12, 4.6, 5);
 2
 3numbers.filter((Number x) -> Integer.class.isInstance(x))
 4    .cast(Integer.class)
 5    .subscribe((Integer x) -> System.out.println(x));
 6// prints:
 7// 1
 8// 7
 9// 12
10// 5

concatMap

作用于Flowable、Observable、Maybe。将数据源的元素作用于指定函数后，将函数的返回值有序的存在新的数据源。

1Observable.range(0, 5)
2    .concatMap(i -> {
3        long delay = Math.round(Math.random() * 2);
4
5        return Observable.timer(delay, TimeUnit.SECONDS).map(n -> i);
6    })
7    .blockingSubscribe(System.out::print);
8
9// prints 01234

concatMapDelayError

与concatMap作用相同，只是将过程发送的所有错误延迟到最后处理。

 1Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS)
 2    .concatMapDelayError(x -> {
 3        if (x.equals(1L)) return Observable.error(new IOException("Something went wrong!"));
 4        else return Observable.just(x, x * x);
 5    })
 6    .blockingSubscribe(
 7        x -> System.out.println("onNext: " + x),
 8        error -> System.out.println("onError: " + error.getMessage()));
 9
10// prints:
11// onNext: 2
12// onNext: 4
13// onNext: 3
14// onNext: 9
15// onError: Something went wrong!

concatMapCompletable

作用于Flowable、Observable。与contactMap类似，不过应用于函数后，返回的是CompletableSource。订阅一次并在所有CompletableSource对象完成时返回一个Completable对象。

 1Observable<Integer> source = Observable.just(2, 1, 3);
 2Completable completable = source.concatMapCompletable(x -> {
 3    return Completable.timer(x, TimeUnit.SECONDS)
 4        .doOnComplete(() -> System.out.println("Info: Processing of item \"" + x + "\" completed"));
 5    });
 6
 7completable.doOnComplete(() -> System.out.println("Info: Processing of all items completed"))
 8    .blockingAwait();
 9
10// prints:
11// Info: Processing of item "2" completed
12// Info: Processing of item "1" completed
13// Info: Processing of item "3" completed
14// Info: Processing of all items completed

concatMapCompletableDelayError

与concatMapCompletable作用相同，只是将过程发送的所有错误延迟到最后处理。

 1Observable<Integer> source = Observable.just(2, 1, 3);
 2Completable completable = source.concatMapCompletableDelayError(x -> {
 3    if (x.equals(2)) {
 4        return Completable.error(new IOException("Processing of item \"" + x + "\" failed!"));
 5    } else {
 6        return Completable.timer(1, TimeUnit.SECONDS)
 7            .doOnComplete(() -> System.out.println("Info: Processing of item \"" + x + "\" completed"));
 8    }
 9});
10
11completable.doOnError(error -> System.out.println("Error: " + error.getMessage()))
12    .onErrorComplete()
13    .blockingAwait();
14
15// prints:
16// Info: Processing of item "1" completed
17// Info: Processing of item "3" completed
18// Error: Processing of item "2" failed!

flatMap

作用于Flowable、Observable、Maybe、Single。与contactMap类似，只是contactMap的数据发射是有序的，而flatMap是无序的。

 1Observable.just("A", "B", "C")
 2    .flatMap(a -> {
 3        return Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS)
 4                .map(b -> '(' + a + ", " + b + ')');
 5    })
 6    .blockingSubscribe(System.out::println);
 7
 8// prints (not necessarily in this order):
 9// (A, 1)
10// (C, 1)
11// (B, 1)
12// (A, 2)
13// (C, 2)
14// (B, 2)
15// (A, 3)
16// (C, 3)
17// (B, 3)

flatMapXXX 和 contactMapXXX

太多了，减少篇幅，大家感兴趣自己查阅官网吧。功能与flatMap和contactMap类似。

flattenAsFlowable & flattenAsObservable

作用于Maybe、Single，将其转化为Flowable，或Observable。

 1Single<Double> source = Single.just(2.0);
 2Flowable<Double> flowable = source.flattenAsFlowable(x -> {
 3    return List.of(x, Math.pow(x, 2), Math.pow(x, 3));
 4});
 5
 6flowable.subscribe(x -> System.out.println("onNext: " + x));
 7
 8// prints:
 9// onNext: 2.0
10// onNext: 4.0
11// onNext: 8.0

groupBy

作用于Flowable、Observable。根据一定的规则对数据源进行分组。

 1Observable<String> animals = Observable.just(
 2    "Tiger", "Elephant", "Cat", "Chameleon", "Frog", "Fish", "Turtle", "Flamingo");
 3
 4animals.groupBy(animal -> animal.charAt(0), String::toUpperCase)
 5    .concatMapSingle(Observable::toList)
 6    .subscribe(System.out::println);
 7
 8// prints:
 9// [TIGER, TURTLE]
10// [ELEPHANT]
11// [CAT, CHAMELEON]
12// [FROG, FISH, FLAMINGO]

scan

作用于Flowable、Observable。对数据进行相关联操作，例如聚合等。

 1Observable.just(5, 3, 8, 1, 7)
 2    .scan(0, (partialSum, x) -> partialSum + x)
 3    .subscribe(System.out::println);
 4
 5// prints:
 6// 0
 7// 5
 8// 8
 9// 16
10// 17
11// 24

window

对数据源发射出来的数据进行收集，按照指定的数量进行分组，以组的形式重新发射。

 1Observable.range(1, 4)
 2    // Create windows containing at most 2 items, and skip 3 items before starting a new window.
 3    .window(2)
 4    .flatMapSingle(window -> {
 5        return window.map(String::valueOf)
 6                .reduce(new StringJoiner(", ", "[", "]"), StringJoiner::add);
 7    })
 8    .subscribe(System.out::println);
 9
10// prints:
11// [1, 2]
12// [3, 4]

4.5

错误处理操作符

onErrorReturn

作用于Flowable、Observable、Maybe、Single。但调用数据源的onError函数后会回到该函数，可对错误进行处理，然后返回值，会调用观察者onNext()继续执行，执行完调用onComplete()函数结束所有事件的发射。

 1Single.just("2A")
 2    .map(v -> Integer.parseInt(v, 10))
 3    .onErrorReturn(error -> {
 4        if (error instanceof NumberFormatException) return 0;
 5        else throw new IllegalArgumentException();
 6    })
 7    .subscribe(
 8        System.out::println,
 9        error -> System.err.println("onError should not be printed!"));
10
11// prints 0

onErrorReturnItem

与onErrorReturn类似，onErrorReturnItem不对错误进行处理，直接返回一个值。

1Single.just("2A")
2    .map(v -> Integer.parseInt(v, 10))
3    .onErrorReturnItem(0)
4    .subscribe(
5        System.out::println,
6        error -> System.err.println("onError should not be printed!"));
7
8// prints 0

onExceptionResumeNext

可作用于Flowable、Observable、Maybe。onErrorReturn发生异常时，回调onComplete()函数后不再往下执行，而onExceptionResumeNext则是要在处理异常的时候返回一个数据源，然后继续执行，如果返回null，则调用观察者的onError()函数。

 1Observable.create((ObservableOnSubscribe<Integer>) e -> {
 2            e.onNext(1);
 3            e.onNext(2);
 4            e.onNext(3);
 5            e.onError(new NullPointerException());
 6            e.onNext(4);
 7        })
 8                .onErrorResumeNext(throwable -> {
 9                    Log.d(TAG, "onErrorResumeNext ");
10                    return Observable.just(4);
11                })
12                .subscribe(new Observer<Integer>() {
13                    @Override
14                    public void onSubscribe(Disposable d) {
15                        Log.d(TAG, "onSubscribe ");
16                    }
17
18                    @Override
19                    public void onNext(Integer integer) {
20                        Log.d(TAG, "onNext " + integer);
21                    }
22
23                    @Override
24                    public void onError(Throwable e) {
25                        Log.d(TAG, "onError ");
26                    }
27
28                    @Override
29                    public void onComplete() {
30                        Log.d(TAG, "onComplete ");
31                    }
32                });

结果：

onExceptionResumeNext操作符也是类似的，只是捕获Exception。

retry

可作用于所有的数据源，当发生错误时，数据源重复发射item，直到没有异常或者达到所指定的次数。

 1boolean first=true;
 2
 3Observable.create((ObservableOnSubscribe<Integer>) e -> {
 4            e.onNext(1);
 5            e.onNext(2);
 6
 7            if (first){
 8                first=false;
 9                e.onError(new NullPointerException());
10
11            }
12
13        })
14                .retry(9)
15                .subscribe(new Observer<Integer>() {
16                    @Override
17                    public void onSubscribe(Disposable d) {
18                        Log.d(TAG, "onSubscribe ");
19                    }
20
21                    @Override
22                    public void onNext(Integer integer) {
23                        Log.d(TAG, "onNext " + integer);
24
25                    }
26
27                    @Override
28                    public void onError(Throwable e) {
29                        Log.d(TAG, "onError ");
30                    }
31
32                    @Override
33                    public void onComplete() {
34                        Log.d(TAG, "onComplete ");
35                    }
36                });

结果：

retryUntil

作用于Flowable、Observable、Maybe。与retry类似，但发生异常时，返回值是false表示继续执行(重复发射数据)，true不再执行,但会调用onError方法。

 1 Observable.create((ObservableOnSubscribe<Integer>) e -> {
 2            e.onNext(1);
 3            e.onNext(2);
 4            e.onError(new NullPointerException());
 5            e.onNext(3);
 6            e.onComplete();
 7        })
 8                .retryUntil(() -> true)
 9                .subscribe(new Observer<Integer>() {
10                    @Override
11                    public void onSubscribe(Disposable d) {
12                        Log.d(TAG, "onSubscribe ");
13                    }
14
15                    @Override
16                    public void onNext(Integer integer) {
17                        Log.d(TAG, "onNext " + integer);
18
19                    }
20
21                    @Override
22                    public void onError(Throwable e) {
23                        Log.d(TAG, "onError ");
24                    }
25
26                    @Override
27                    public void onComplete() {
28                        Log.d(TAG, "onComplete ");
29                    }
30                });

结果：

retryWhen与此类似，但其判断标准不是BooleanSupplier对象的getAsBoolean()函数的返回值。而是返回的 Observable或Flowable是否会发射异常事件。

总结

太多操作符太累了，看得心好累。还是根据实际开发需要查阅文档才是正确的姿势。本文只是RxJava冰山一角，更多请参阅官网。同时不建议立马在项目上实践，给它点时间报bug。

参阅官网:

https://github.com/ReactiveX/RxJava

好东西要分享:

https://github.com/GitCode8/GitCode/blob/master/README.md

如果你看到了这，点个赞，收下我双膝。如果文章有误，帮忙指正，谢谢大佬们。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2019-07-09，如有侵权请联系 cloudcommunity@tencent.com 删除

android

编程算法

本文分享自 Android补给站微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

android

编程算法