欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Android开发——关于RxJava的知识总结

程序员文章站 2022-06-11 14:26:20
...

0. 前言

RxJava主页上的介绍:

//a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
//一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。

大家都知道异步代码经常会既难写也难被读懂。这时候RxJava的优势就来了,随着程序逻辑变得越来越复杂,它依然能够保持简洁。初学RxJava只要把握两点,观察者模式和异步

 

1. 观察者模式

被观察者:被监视的对象,当某个状态改变的时候会通知观察者;

观察者:监视着被观察者的行为,当被观察者某个状态改变的时候会通知观察者,观察者会执行对应的操作

订阅:将观察者和被观察者建立联系。

这里举一个Android中常见的例子,点击Button后触发OnClickListener中的onClick()事件。在这个例子中被观察者就是Button、观察者:OnClickListener、订阅的逻辑就是setOnClickListener的过程。

 

2. RxJava

2.1 RxJava的简单示例

RxJava也是基于观察者模式来组建自己的程序逻辑的,就是构建被观察者(Observable),观察者(Observer/Subscriber,然后通过Subscribe()建立两者的订阅关系。

直接看代码比较容易懂:

//功能为按顺序打印一些字符串
//别忘记添加依赖compile 'io.reactivex:rxjava:1.0.14'   compile 'io.reactivex:rxandroid:1.0.1'

//创建一个观察者
Observer<String> observer = new Observer<String>() {

            @Override
            public void onCompleted() {
                Log.i(TAG, "Completed");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "Error");
            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, s);
            }
};

//使用Observable.create()创建被观察者
Observable observable1 = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello");
                subscriber.onNext("Wrold");
                subscriber.onCompleted();
            }
});
//订阅
observable1.subscribe(observer);

2.2  一些拓展

2.2.1  关于Observable的简单写法

上面介绍了Observable的一种最正规的创建方式,当然也可以偷懒写成下面这三种的样子,效果是一样的:

Observable observable2 = Observable.just("Hello", "World");

String [] words = {"Hello", "World"};
Observable observable3 = Observable.from(words);

List<String> list = new ArrayList<String>();
list.add("Hellow");
list.add("Wrold");
Observable observable4 = Observable.from(list);

2.2.2  关于Observer的扩展

除了Observer接口之外,RxJava 还内置了一个实现了Observer的抽象:Subscriber

SubscriberObserver接口进行了一些扩展,但他们的基本使用方式是完全一样的。而且不仅基本使用方式一样,实质上在 RxJava subscribe 过程中,Observer也总是会先被转换成一个Subscriber再使用不仅基本使用方式一样,实质上在 RxJava subscribe 过程中,Observer也总是会先被转换成一个Subscriber再使用

Subscriber<String> subscriber = new Subscriber<String>() {
  @Override
  public void onNext(String s) {
      Log.d(tag, "Item: " + s);
  }

  @Override
  public void onCompleted() {
      Log.d(tag, "Completed!");
  }

  @Override
  public void onError(Throwable e) {
      Log.d(tag, "Error!");
  }
};

Android开发——关于RxJava的知识总结


它们的区别对于使用者来说主要有两点:

1onStart(): 这是Subscriber增加的方法。它会 subscribe 刚开始,而事件还未发送之前被调用

onStart()可以用于做一些准备工作,例如数据的清零或重置。这是一个可选方法,默认情况下它的实现为空。需要注意的是,onStart()总是在 subscribe 所发生的线程被调用,而不能指定线程。因此主线程中UI的准备逻辑,就不适宜在这里写了。

(2)unsubscribe(): 这是Subscriber所实现的另一个接口Subscription的方法,用于取消订阅

在这个方法被调用后,Subscriber将不再接收事件。一般在这个方法调用前,可以使用isUnsubscribed()先判断一下状态。unsubscribe()这个方法很重要,可以较少内存泄露的风险。因为在subscribe()之后,Observable会持有Subscriber的引用,这个引用如果不能及时被释放,将有内存泄露的风险。

 

2.2  Action

有时候我们不需要onErroronCompleted这两个方法的观察者回调,而只需要onNext()方法,这时就可以使用Action来代替Subscriber。如下所示:

Observable.just("Hello", "World").subscribe(new Action1<String>() {
      @Override
      public void call(String s) {
            Log.i(TAG, s);
      }
});


2.2.1  什么是Action

Action的使用为我们减少了不必要的代码,使得写出的代码看上去更加得简洁。ActionRxJava 的一个接口,常用的有Action0Action1

(1)Action0 它只有一个方法 call(),这个方法是无参无返回值的;由于 onCompleted() 方法也是无参无返回值的,因此 Action0 可以被当成一个包装对象,将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。
2Ation1:它同样只有一个方法 call(T param)这个方法也无返回值,但有一个参数;与 Action0 同理,由于 onNext(T obj) onError(Throwable error) 也是单参数无返回值的,因此 Action1 可以将 onNext(obj)onError(error) 打包起来传入 subscribe() 以实现不完整定义的回调

Observable observable = Observable.just("Hello", "World");
      //处理onNext()中的内容
      Action1<String> onNextAction = new Action1<String>() {
          @Override
          public void call(String s) {
              Log.i(TAG, s);
          }
      };
      //处理onError()中的内容
      Action1<Throwable> onErrorAction = new Action1<Throwable>() {
          @Override
          public void call(Throwable throwable) {

          }
      };
      //处理onCompleted()中的内容
      Action0 onCompletedAction = new Action0() {
          @Override
          public void call() {
              Log.i(TAG, "Completed");

          }
      };

//使用 onNextAction 来定义 onNext()
Observable.just("Hello", "World").subscribe(onNextAction);
//使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
Observable.just("Hello", "World").subscribe(onNextAction, onErrorAction);
//使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
Observable.just("Hello", "World").subscribe(onNextAction, onErrorAction, onCompletedAction);

//现在有个疑问,为什么使用Action也能达到使用Subscriber的结果?
//进subscribe(Action1 onNext)的源码看看。如下图所示
//原来就是把Action对象转化成对应的Subscriber对象了
//这样就不难理解为什么可以使用Action来代替Subscriber了。

Android开发——关于RxJava的知识总结


2.3  map/flatMap

在使用map之前还要说一个接口:Func1。

Func1和上一篇提到的Action1相似。Func1和Action1的区别在于,Func1包装的是有返回值的方法。

接下来就是map的用法,看代码更直观一点。

//得到多个Student对象中的name,保存到nameList中

Observable.just(student1, student2, student2)
//使用map进行转换,参数1:转换前的类型,参数2:转换后的类型
.map(new Func1<Student, String>() {
       @Override
       public String call(Student i) {
             String name = i.getName();
             return name;
       }
})
.subscribe(new Action1<String>() {
      @Override
      public void call(String s) {
           nameList.add(s);
      }
});
可以看到Observable中原来的参数是Student对象,而最后我们需要的是name,这里使用了map来实现这一转换的过程。当然,map可以多次被使用。

flatMap是一个比较难理解的一个转换,在这里先假设一个需求,需要打印多个Student所学的课程。这跟之前获取Student的name又不同了,一个Student类中只有一个name,但是一个Student却有多门课程Course。

     * 学生类
     */
    class Student {
        private String name;//姓名
        private List<Course> coursesList;//所修的课程
        ...
    }
    /**
     * 课程类
     */
    class  Course {
        private String name;//课程名
        private String id;
        ...
}
如果用map来实现打印所有学生所修课程名,实现代码是这样的:

        List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Action1<List<Course>> action1 = new Action1<List<Course>>() {
            @Override
            public void call(List<Course> courses) {
                //遍历courses,输出cuouses的name
                 for (int i = 0; i < courses.size(); i++){
                    Log.i(TAG, courses.get(i).getName());
                }
            }
        };
        Observable.from(students)
                .map(new Func1<Student, List<Course>>() {
                    @Override
                    public List<Course> call(Student student) {
                        //返回coursesList
                        return student.getCoursesList();
                    }
                })
                .subscribe(action1);
可以看到在Action1中出现了for循环来打印课程名,如果我们为了代码的复用性,希望Subscriber中直接传入单个Course对象呢,用map这种一对一的转换显然就不行了,我们现在就可以使用flatMap实现一对多的转换。

        List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Observable.from(students)
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        return Observable.from(student.getCoursesList());
                    }
                })
                .subscribe(new Action1<Course>() {
                    @Override
                    public void call(Course course) {
                        Log.i(TAG, course.getName());
                    }
                });
flatMap的原理是这样的:

(1)使用传入的事件对象创建一个Observable对象;

(2)并不发送这个Observable,而是**它,使其开始发送事件;

(3)每一个创建出来的Observable发送的事件,都被汇入到同一个Observab,而这个Observable负责将这些事件统一交给Subscriber的回调方法。

经过了这三步,就把事件拆分成了两级,通过一组新创建的Observable将初始的对象铺平之后通过统一路径分发了下去。


关于RxJava的其他操作符,比如filter,take,skip等,有兴趣的可以在这个训练项目中进行熟悉练习。


2.4  线程控制

RxJava在不指定线程的情况下,发起时间和消费时间默认使用当前线程。因此若是map中有耗时的操作,都是在主线程中进行的,这样会导致主线程拥塞。

Scheduler:线程控制器,就是来解决上述问题的,它可以指定每一段代码在什么样的线程中执行。

模拟一个需求:新的线程发起事件,在主线程中消费,这就用到了subscribeOn()和observeOn()方法来指定发生的线程和消费的线程。前者指定事件产生的线程,只能指定一次,如果指定多次则以第一次为准。后者指定Subscriber所运行的线程,或者说是事件消费的线程,可以指定多次,每次指定完都会在下一步一直生效。下面是这个需求的实现代码,以及关于这两个方法参数,RxJava为我们提供的几个Scheduler。

Observable.just("Hello", "Word")
      .subscribeOn(Schedulers.newThread())//指定 subscribe() 发生在新的线程
      .observeOn(AndroidSchedulers.mainThread())// 指定 Subscriber 的回调发生在主线程
      .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, s);
                    }
       });

//1.Schedulers.immediate():直接在当前线程运行,相当于不指定线程。这是默认的 Scheduler。
//2.Schedulers.newThread():总是启用新线程,并在新线程执行操作。
//3.Schedulers.io(): I/O 操作(读写文件、读写数据库、网络信息交互等)所使用的 Scheduler。
//和 newThread() 区别在于 io() 的内部实现是是用一个无数量上限的线程池,可以重用空闲的线程
//因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中,避免创建不必要的线程。
//4.Schedulers.computation():计算所使用的 Scheduler。
//这个计算指的是 CPU 密集型计算,即不会被 I/O 等操作限制性能的操作,例如图形的计算。
//使用CPU核数大小的固定线程池,I/O 操作如果被放在这里,会因为I/O等待的时间而浪费 CPU。
//5.AndroidSchedulers.mainThread():它指定的操作将在 Android 主线程运行。

2.5  多次切换线程

上面只是对事件的发起和消费制定了线程。如果中间有map之类的操作呢?是否可以实现发起的线程在新线程中,map的处理在IO线程,最后的消费在主线程中呢?

可以看到observeOn()被调用了两次,分别指定了map的处理的现场和消费事件show(s)的线程。

        Observable.just("Hello", "Wrold")
                .subscribeOn(Schedulers.newThread())//指定在新的线程中发起
                .observeOn(Schedulers.io())         //指定在io线程中处理
                .map(new Func1<String, String>() {
                    @Override
                    public String call(String s) {
                        return handleString(s);       //处理数据
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())//指定在主线程中处理
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        show(s);                       //消费事件
                    }
                });



参考文献:

http://gank.io/post/560e15be2dca930e00da1083

http://www.jianshu.com/p/19cac3c5b106