★42.RxJava2

简介

  • 相关网站：GitHub地址、官方文档、中文文档。
  • RxJava2 通过一种扩展的 观察者模式 来实现 异步 执行任务。
  • RxJava2 能方便地进行 线程 切换。
  • RxJava2 能方便地把 异步 执行的代码写在一处。

基础知识

相关概念

  • Observable： 被观察者 。
  • Observer： 观察者 。
  • Subscriber： 订阅者 ，实现了Observer接口，多了unsubscribe()，用来取消订阅。
  • Subscription：类似Subscriber，Observable调用subscribe()方法返回的对象。
  • Subject：可以当作Observable或Observer来用。
  • subscribe()： 订阅 方法。
  • Event：事件。
  • Scheduler： 调度器 ，相当于 线程控制器 。
  • Action0：接口，里面只有一个无返回值0参数的call()，同理有Action1、Action2等，代表着1个、2个参数等。
  • Func0：接口，类似Action0，区别是Func0有返回值。

相关方法

  • onNext()：在 事件队列 中，进入下一个事件时调用，同时也是 事件处理方法 。
  • onCompleted()： 事件队列 完成时调用。
  • onError()： 事件队列 错误时调用。

简单示例

1. 普通示例

1. 定义Observable 

Observable<Integer> observable = Observable.create(emitter -> {
    // Todo: 执行后台请求，请求后的结果通过onNext()发送给Observer
    // 发送消息
    emitter.onNext(1);
    emitter.onNext(2);
    emitter.onNext(3);
    // 消息发送完毕
    emitter.onComplete();
});

2. 定义Observer 

Observer<Integer> observer = new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "subscribe");
        // 订阅时
        // Disposable对象可以保存，日后通过调用Disposable.dispose()来中断订阅。
    }

    @Override
    public void onNext(Integer value) {
        Log.d(TAG, "" + value);
        // 接收到事件时
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "error");
        // 接收到错误时
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "complete");
        // 事件接收完毕时
    }
};

3. 订阅

注意 ：此处代码的书写顺序看起来不符合直觉，这样设计是为了便于 链式调用 。

observable.subscribe(observer);

2. 链式示例

Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
    // Todo: 执行后台请求，请求后的结果通过onNext()发送给Observer
    // 发送消息
    emitter.onNext(1);
    emitter.onNext(2);
    emitter.onNext(3);
    // 消息发送完毕
    emitter.onComplete();
}).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "subscribe");
        // 订阅时
        // Disposable对象可以保存，日后通过调用Disposable.dispose()来中断订阅。
    }

    @Override
    public void onNext(Integer value) {
        Log.d(TAG, "" + value);
        // 接收到事件时
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "error");
        // 接收到错误时
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "complete");
        // 事件接收完毕时
    }
});

3. 无Observer示例

subscribe()的所有重载形式

public final Disposable subscribe() {}
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}

代码

Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
    emitter.onNext(1);
    emitter.onNext(2);
    emitter.onNext(3);
    emitter.onComplete();
    emitter.onNext(4);
}).subscribe(
        integer -> Log.d(TAG, "onNext: " + integer),
        throwable -> Log.d(TAG, "onError"),
        () -> Log.d(TAG, "onComplete"));

示例解说

  • emitter是ObservableEmitter类型，用于发射onNext()，onComplete()，onError()消息。
  • Disposable对象可以保存，日后通过调用Disposable.dispose()来 中断订阅 ， 中断订阅 以后发送的消息无法被Observer接收到。
  • Observable可以不发送onComplete()或onError()。
  • 当Observable发送了一个onComplete()后，Observable在onComplete()之后的onNext()将会继续发送，而Observer收到onComplete()之后将不再继续接收onNext()。
  • 当Observable发送了一个onError()后，Observable在onError()之后的onNext()将继续发送，而Observer收到onError()之后将不再继续接收onNext()。
  • onComplete()和onError()必须 唯一 并且 互斥 ，即只能发送一个onComplete()或onError()。

事件流向图

事件流向 示意图
只发送onNext()事件
发送onComplete()事件
发送onError()事件

线程相关

切换线程示例

  • subscribeOn()：控制observable在什么 线程 中发送事件。
  • observeOn()：控制observer在什么 线程 中处理事件。
// observable的链式操作中
observable.subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(observer);

线程种类

线程种类 描述
Schedulers.newThread() 新 线程 。
Schedulers.computation() 用于CPU计算密集型的操作的 线程 。
Schedulers.io() 用于IO操作的 线程 ，如网络IO，文件IO，数据库IO。
AndroidSchedulers.mainThread() Android 的 UI主线程 。

注意事项

  • Observable和Observer默认是在 UI主线程 中运行的。
  • Observable多次切换线程的话，只有第一次有效。
  • Observer多次切换线程的话，只有最后一次生效。

过滤操作符

Sample

简介

  • Sample操作符定时查看一个Observable，然后发射自上次采样以来它最近发射的数据。
  • 会导致某些事件丢失。

示意图

变换操作符

Map

示意图

简介

用于将一种类型的 事件 转换为另一种类型的 事件 。

啰嗦示例

Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
    emitter.onNext(1);
}).map(new Function<Integer, String>() {
    @Override
    public String apply(@NonNull Integer integer) throws Exception {
        return "This is result " + integer;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.d(TAG, s);
    }
});

简洁示例

Observable.create((ObservableOnSubscribe<Integer>) emitter -> emitter.onNext(1))
        .map(integer -> "This is result " + integer)
        .subscribe(s -> Log.d(TAG, s));

FlatMap

示意图

简介

  • FlatMap将单个 事件 或 事件队列 变换为一个发送单个 事件 或 事件队列 的Observable。
  • 可以通过FlatMap将 事件 变换为Observable来实现链式中连接多个Observable： Observable->事件->Observable->事件...->Observer 。
  • FlatMap并不保证 事件 的顺序，如果需要保证顺序则使用ConcatMap，使用方法同FlatMap。

啰嗦示例

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        emitter.onNext(1);
    }
}).flatMap(new Function<Integer, Observable<String>>() {
    @Override
    public Observable<String> apply(Integer integer) throws Exception {
        final List<String> list = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            list.add("I am value " + integer);
        }
        return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.d(TAG, s);
    }
});

简洁示例

Observable.create((ObservableOnSubscribe<Integer>) emitter -> emitter.onNext(1))
        .flatMap(new Function<Integer, Observable<String>>() {
            @Override
            public Observable<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        })
        .subscribe(s -> Log.d(TAG, s));

ConcatMap

  • ConcatMap保证 事件 的顺序，使用方法同FlatMap。
  • 示例省略，参考FlatMap。

Zip

示意图

简介

  • 组合多个Observable发送的 事件 ，然后发送这个 事件组合 。
  • 事件组合 的顺序是严格按照 事件 发送的顺序来进行的。
  • 发送 事件 的数量，与所有Observable中 事件 数量最少的那个一样。
  • 注意：这多个Observable在同一个 线程 时无法将 事件 组合发送，即发送一个Observable的所有 事件 以后，再发送另一个Observable。
  • 应用场景：一个界面需要展示用户的一些信息，而这些信息分别要从两个服务器接口中获取，而只有当两个都获取到了之后才能进行展示。

啰嗦示例

Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        Log.d(TAG, "emit 1");
        emitter.onNext(1);

        Log.d(TAG, "emit complete1");
        emitter.onComplete();
    }
}).subscribeOn(Schedulers.io());

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        Log.d(TAG, "emit A");
        emitter.onNext("A");

        Log.d(TAG, "emit complete2");
        emitter.onComplete();
    }
}).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
    @Override
    public String apply(@NonNull Integer integer, @NonNull String s) throws Exception {
        return integer + s;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.d(TAG, "onNext: " + value);
    }
});

简洁示例

Observable<Integer> observable1 = Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
    Log.d(TAG, "emit 1");
    emitter.onNext(1);

    Log.d(TAG, "emit complete1");
    emitter.onComplete();
}).subscribeOn(Schedulers.io());

Observable<String> observable2 = Observable.create((ObservableOnSubscribe<String>) emitter -> {
    Log.d(TAG, "emit A");
    emitter.onNext("A");

    Log.d(TAG, "emit complete2");
    emitter.onComplete();
}).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2, (integer, s) -> integer + s).subscribe(s -> Log.d(TAG, "onNext: " + s));

特殊情况处理

对于Observer在 UI主线程 进行 事件 处理，但是接收到 事件 时，所在的Activity或Fragment已经退出的情况，需要通过调用Disposable.dispose()或Subscription.cancel()来 中断订阅 。

1. 创建CompositeDisposable

  • CompositeDisposable对象是储存Disposable对象的容器。
  • 作为Activity或Fragment的成员： 
    CompositeDisposable mCompositeDisposable = new CompositeDisposable();

2. 保存Disposable

在Observer.onSubscribe(...)中：

@Override
public void onSubscribe(Disposable d) {
    mCompositeDisposable.add(d);
}

3. 中断订阅

在Activity或Fragment的onDestroy()中：

mCompositeDisposable.clear();

Flowable

简介

  • Flowable类似Observable，但是性能不如Observable。
  • Flowable为解决 Backpressure问题 而生。

Backpressure问题简介

  • Backpressure问题 ：当 事件 的发送远远快于 事件 的消耗时，未消耗的 事件 会堆积起来，最终可能发生 OOM 。
  • 出现 Backpressure问题 可能是因为两个原因：
    • 某个Observable发送 事件 的速度太快，或数量太多。
    • 某个Observer处理 事件 的速度太慢。
  • 事件 的发送和接收在同一个 线程 的时候不会出现这种问题，因为必定会处理完一个 事件 以后才能继续发送下一个 事件 。

简洁示例

Flowable<Integer> upstream = Flowable.create(emitter -> {
    // 直到下游开始请求事件
    while (emitter.requested() == 0) {
        if (emitter.isCancelled())
            break;
    }
    Log.d(TAG, "emit 1");
    emitter.onNext(1);
    Log.d(TAG, "emit complete");
    emitter.onComplete();
}, BackpressureStrategy.ERROR);

Subscriber<Integer> downstream = new Subscriber<Integer>() {
    @Override
    public void onSubscribe(Subscription s) {
        Log.d(TAG, "onSubscribe");
        s.request(10);
    }

    @Override
    public void onNext(Integer integer) {
        Log.d(TAG, "onNext: " + integer);
    }

    @Override
    public void onError(Throwable t) {
        Log.w(TAG, "onError: ", t);
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "onComplete");
    }
};

示例解说

  • BackpressureStrategy（ 背压策略 ），表示 事件 发送速度快于消耗速度时说使用的策略，有以下几种：
    • MISSING：多出来的 事件 直接丢弃。
    • ERROR：抛出MissingBackpressureException异常。
    • BUFFER：全部 事件 缓存直到被消耗，注意 OOM 问题。也可以通过Flowable.onBackpressureBuffer()设置。
    • DROP：丢弃最近发送的 事件 。也可以通过Flowable.onBackpressureDrop()设置。
    • LATEST：只保留一个最近的 事件 覆盖前一个 事件 。也可以通过Flowable.onBackpressureLatest()设置。
  • Subscription.cancel()：可以用于 取消订阅关系 。
  • Subscription.request(..)：告知Flowable，Observer的 事件请求量 ，可以多次调用，每次调用累加 事件请求量 计数器。
  • FlowableEmitter.requested()：得知Observer的 事件请求量 。
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容