RxJava学习总结

Posted AC_Jobim

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RxJava学习总结相关的知识，希望对你有一定的参考价值。

这里写目录标题

参考博客：RxJava2最全面、最详细的讲解（一）

一、RxJava概述

在这里插入图片描述

RxJava最核心的两个东西Observable（被观察者、事件源）和Observer（观察者），Observable发出一系列的事件，Observer处理这些事件。在Observer接收到事件处理之前我们很方便地对结果做出各种拦截处理等。

RxJava的流程图大致如下：

img

添加依赖：

implementation 'io.reactivex.rxjava2:rxjava:2.0.4'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

二、RxJava的使用

2.1 RxJava观察者模式

RxJava有四个基本概念：Observer(观察者)，Observable(被观察者)，subscribe(订阅)，事件。Observer和Observable通过subscribe()实现订阅关系，从而Observable可以在需要的时候发出事件通知Observer。

  • Observer： 观察者，它决定事件发生时有怎么样的行为；
  • Observable： 被观察者，它决定什么时候出发事件以及触发什么样的事件；
  • subscribe： 订阅，将Observer和Observable关联起来。

2.2 RxJava的基本实现

1、创建被观察者Observable

它决定什么时候出发事件以及触发怎么样的事件，通过Observable.create(ObservableOnSubscribe)创建被观察者实例，这是最基本的创造事件序列的方法，并为它定义事件出发规则。

//创建被观察者Observable
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        Log.e(TAG, "subscribe()线程==" + Thread.currentThread().getName());
        e.onNext("RxJava：e.onNext== 第一次");
        e.onNext("RxJava：e.onNext== 第二次");
        e.onNext("RxJava：e.onNext== 第三次");
        e.onComplete();
    }
});

创建一个ObservableOnSubscribe对象并且实现subscribe()方法，设定了事件的内容是String，并返回ObservableEmitter，相当于一个计划表，当Observable被订阅的时候，复写subscribe()方法定义发送的事件，ObservableEmitter是事件发射器，定义并且向观察者发送需要发送的事件，onNext()会被执行三次，最后执行onComplete()方法，这样由被观察者调用观察者回调的方法，实现了被观察者向观察者传递事件。

2、创建观察者Observer

它决定事件触发有怎样的行为，定义响应事件的行为，直接new一个Observer观察者实例，实现其中相应的方法：

//创建观察者Observer
Observer<String> observer = new Observer<String>() {
     @Override
     public void onSubscribe(Disposable d) {
         Log.e(TAG, "onSubscribe == 订阅,"+ Thread.currentThread().getName());
     }

     @Override
     public void onNext(String s) {
         Log.e(TAG, "onNext == " + s+","+Thread.currentThread().getName());
     }

     @Override
     public void onError(Throwable e) {
         Log.e(TAG, "onError == " + e.getMessage());
     }

     @Override
     public void onComplete() {
         Log.e(TAG, "onComplete ,"+Thread.currentThread().getName());
     }
 };
  • onSubscribe(Disposable d)： 事件订阅成功回调，返回Disposable请求实例，可以通过d.dispose()取消请求；
  • onNext(T)： 响应事件的方法，发送事件时，观察者会回调onNext()方法，接收事件数据；
  • onError()： 事件队列异常，在处理事件出现异常的时候回触发这个方法，其他事件不会再继续发出；
  • onComplete()： 事件队列完结，当不再有新onNext()发出时，需要触发onComplete()方法来作为标志，其他事件不会再继续发出

注意：在一个正确的事件运行队列中，onError()和onComplete()有且仅有一个出现，还是在事件的最后出现，即onError()和onComplete()是互斥的，当一个出现了，另一个就不再回出现。

3、subscribe()

订阅，连接Observable(被观察者)和Observer(观察者)。

 //订阅(观察者观察被观察者)
 observable.subscribe(observer);

Observable是被观察者，observer是观察者，创建完Observer和Observable之后，通过subscribe()将两者关联起来。

通过该调用，回调观察者的相关方法，从而响应被观察者响应的事件，Observable只生产事件，真正发送事件的是在它订阅的时候，即subscribe()被调用的时候。

执行结果：
在这里插入图片描述

另外：观察者Obaserver的subscribe具有多个重载的方法：

    //观察者不对被观察者发送的事件做出响应(但是被观察者还可以继续发送事件)
    public final Disposable subscribe()
 
    //观察者对被观察者发送的任何事件都做出响应
    public final void subscribe(Observer<? super T> observer)
 
    //表示观察者只对被观察者发送的Next事件做出响应
    public final Disposable subscribe(Consumer<? super T> onNext)
 
    //表示观察者只对被观察者发送的Next & Error事件做出响应
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
 
    //表示观察者只对被观察者发送的Next & Error & Complete事件做出响应
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                      Action onComplete)
 
    //表示观察者只对被观察者发送的Next & Error & Complete & onSubscribe事件做出响应
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                      Action onComplete, Consumer<? super Disposable> onSubscribe)

2.3 Scheduler调度者

在RxJava默认规则中，事件的发出和消费都是在同一个线程中发生的，那么上面的例子来说，就是一个同步的观察者模式。观察者模式的本省就是后台处理，前台回调的异步机制，因此异步对RxJava来说是至关重要的，异步的实现则需要用到Scheduler调度器来切换线程。

在RxJava中Scheduler(调度器)相当于线程控制器，RxJava通过Scheduler来指定那一部分代码执行在哪一个线程。我们来看看简单的例子：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        Log.e(TAG, "subscribe()线程==" + Thread.currentThread().getName());
        e.onNext("RxJava：e.onNext== 第一次");
        e.onNext("RxJava：e.onNext== 第二次");
        e.onNext("RxJava：e.onNext== 第三次");
        e.onComplete();
    }
}).subscribeOn(Schedulers.io())//指定被观察者subscribe()(发送事件的线程)在IO线程()
        .observeOn(AndroidSchedulers.mainThread());//指定观察者接收响应事件的线程在主线程

log如下：
在这里插入图片描述

执行事件subscribe()方法的线程为IO线程，回调事件方法onNext()的线程为主线程。

由于subscribeOn(Schedulers.io())指定了subscribe()方法发送事件线程在IO线程中执行，observeOn(AndroidSchedulers.mainThread())指定了接收事件在主线程中执行。这种方式非常常见，适用于后台获取数据，前台显示的程序策略。

  • subscribeOn()： 指定Observable被观察者subscribe()时所发生的线程，即指定发生事件的线程
  • observeOn()： 指定Observer观察者接收&响应事件的线程，即订阅者接收事件的线程

注意：多次指定发射事件的线程只有第一次指定有效，也就是说多次调用subscribeOn()只有第一次有效，其余的会被忽略；但是多次指定订阅者接收事件的线程是可以的，也就是说每observeOn()一次，接收事件的线程就会切换一次。

RxJava中内置了很多线程项供我们选择：

  • Schedulers.io()： 代表IO操作的线程，通常用于网络、读写文件等IO密集型的操作。行为模式和new Thread()差不多，只是IO的内部是一个无上限的线程池，可重用空闲的线程，更高效（不要把计算工作放在IO内，可以避免创建不必要的线程）；
  • AndroidSchedulers.mainThread()：Android的主线程；用于更新UI
  • Schedulers.newThread()： 总是启用新线程，并在新线程中执行操作；多用于耗时操作
  • Schedulers.computation()： 代表CPU计算密集型的操作，即不会被IO等操作限制性能的操作。

三、操作符介绍

3.1 创建操作符

几种常用的创建操作符：

方法作用备注使用场景
基本创建create()创建一个完成的被观察者(Observable)RxJava中创建被观察者最基本的操作符1、完整&快速创建被观察者2、数组、集合遍历
快速创建empty()快速创建后只发送complete事件，直接通知完成同上
同上error()快速创建后只发送error事件，直接通知异常同上
同上never()快速创建后不发送任何事件同上
同上just()快速创建后直接发送传入的事件参数最多只能10个，即发送的事件最多10个同上
同上fromArray()快速创建后直接发送传入的数组数据参数为数组，子类型为任意类型，可10个以上同上
同上fromIterable()快速创建后直接发送传入的集合List数据参数为集合List，子类型为任意类型，可10个以上同上
延时创建defer()直到观察者Observer订阅被观察者Observable时，才动态创建被观察者&发送事件通过Observable工厂方法创建被观察者，每次订阅后都会得到一个最新创建的被观察者Observable，确保里面的数据是最新的1、定时操作2、周期性操作
同上timer()快速创建被观察者，指定延时时间，发送一个数值为0的事件延时指定时间后发送一个参数为0的事件，相当于onNext（0）同上
同上interval()快速创建被观察者，每隔指定时间发送事件发送事件序列，从0开始，无限递增1同上
同上intervalRange()快速创建被观察者，每隔指定时间发送事件，可指定发送事件数发送无限递增1的事件序列，可指定起始值大小和事件次数，可延时，类似interval()同上
同上range()快速创建被观察者，连续发送一个时间序列，可指定范围发送无限递增1的事件序列，可指定起始值大小和事件次数，不可延时，类似intervalRange()同上
同上rangeLong()同上，区别在于数据参数为类型Long同上同上

3.1.1 快速创建

1、create()：是所有创建型操作符的“根”，也就是说其他创建型操作符最后都是通过create()来创建的Observable的。

Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> emitter) throws Exception {
            try{
                if (!emitter.isDisposed()){
                    emitter.onNext("RxJava：e.onNext== 第一次");
                    emitter.onNext("RxJava：e.onNext== 第二次");
                    emitter.onNext("RxJava：e.onNext== 第三次");
                    emitter.onComplete();
                }
            }catch (Exception e){
                emitter.onError(e);
            }
        }
    }).subscribe(new Observer<String>() {
        //默认最先复写onSubscribe()
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(String s) {
            Log.e(TAG, "onNext == " + s);
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "onError == " + e.getMessage());
        }
 
        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete == ");
        }
    });

在使用create()操作符时，最好在被观察者的回调函数subscribe()中加上isDisposed()，以便在观察者断开连接的时候不在执行subscribe()函数中的相关逻辑，避免意想不到的错误出现。

打印数据如下：
img

2、empty()：快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。

//快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。
Observable.empty()
           .subscribe(new Observer<Object>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "empty：onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(Object o) {
            Log.e(TAG, "empty：onNext ==" + o.toString());
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "empty：onError == " + e.getMessage());
        }
 
        @Override
        public void onComplete() {
            Log.e(TAG, "empty：onComplete == ");
        }
    });

打印log如下：
img

3、error()：快速创建被观察者对象，仅发送onError()事件，直接通知异常。

    //快速创建被观察者对象，仅发送onError()事件，直接通知异常。
    Observable.error(new Throwable("只回调error"))
            .subscribe(new Observer<Object>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "error：onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(Object o) {
            Log.e(TAG, "error：onNext ==" + o.toString());
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "error：onError == " + e.getMessage());
        }
 
        @Override
        public void onComplete() {
            Log.e(TAG, "error：onComplete == ");
        }
    });

打印log如下：

img

4、never()：快速创建被观察者对象，不发送任何事件。

    //快速创建被观察者对象，不发送任何事件。
    Observable.never()
              .subscribe(new Observer<Object>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "never：onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(Object o) {
            Log.e(TAG, "never：onNext ==" + o.toString());
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "never：onError == " + e.getMessage());
        }
 
        @Override
        public void onComplete() {
            Log.e(TAG, "never：onComplete == ");
        }
    });

打印log如下：

img

5、Just(T… items)：快速创建被观察者对象，最多只能发送10个事件。

  • 参数类型：任意类型
Observable.just(1, 2, 3, 4, 5).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "just：onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(Integer integer) {
            Log.e(TAG, "just：onNext == " + integer);
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "just：onError == " + e.getMessage());
        }
 
        @Override
        public void onComplete() {
            Log.e(TAG, "just：onComplete == ");
        }
    });

通过just()创建传入Integer类型的参数构建Observable被观察者，相当于执行了onNext(1)~onNext(5)，通过链式编程订阅观察者。log如下：

img

6、fromArray(T… items)：快速创建被观察者对象，可发送多个任意子类型的数据。

  • 参数类型：数组，子类型可以为任意类型
    //设置需要传入的数组
    String[] strings = {"商品类","非商品类"};
    //传入数组，被观察者创建后会将数组转换成Observable并且发送里面所以的数据
    Observable.fromArray(strings).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e(TAG, "fromArray：onSubscribe == 订阅");
        }
 
        @Override
        public void onNext(String s) {
            Log.e(TAG, "fromArray：onNext == " + s);
        }
 
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "fromArray：onError == " + e.getMessage());
        }
 
   
以上是关于RxJava学习总结的主要内容，如果未能解决你的问题，请参考以下文章
 

 学习 rxjava2 在android 上的应用总结


 原创专栏RxJava源代码剖析


 Python学习总结


 Android - 框架之Retrofit+RxJava的使用


 当Kotlin遇见RxJava多数据源


 使用 RxJava 将数据从 ViewModel 移动到 Fragment