Android框架第（三）篇--响应式编程

2017-11-05 本文已影响39人 ZebraWei

RxJava响应式编程

上篇简单的阐述了响应式编程的基本理论。这篇主要对响应编程进行详细的介绍。以及主要介绍响应编程的主要实现-RxJava。其中RxJava的操作符和调度器(Scheduler)主要介绍对象。

一、什么是响应式编程？

响应式编程：是一种基于异步数据流概念的编程模式。关键概念：事件把现实中的事件，搬到软件中区去，事件过程等待触发到另一种事件。响应式编程最通用的一个场景是UI：我们的移动App必须做出对网络调用、用户触摸输入和系统弹框的响应。RxJava是响应式编程的具体实现，是异步数据处理库扩展的观察者模式。

二、扩展观察者模式

扩展的观察者模式：有四个基本的要素：Observable (被观察者)、 Observer (观察者)、 subscribe (订阅)、event(事件)。Observable (被观察者)和 Observer (观察者)通过 subscribe() 方法实现订阅关系，Observable就可以在需要的时候通知Observer。其中onCompleted、onError、onNext是必须要实现的方法，他们的含义分别是：

1.onCompleted：事件队列完结，RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。当不会再有新的 onNext发出时，需要触发 onCompleted() 方法作为完成标志。
2.onError：事件队列异常，在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
3.onNext：普通的事件，将要处理的事件添加到事件队列中。
4.onStart：它会在事件还未发送之前被调用，可以用于做一些准备工作。例如数据的清零或重置，这是一个可选方法，默认情况下它的实现为空。

三、基本概念

RxJava的四个基本角色：Observable、Observer、Subscriber、Subject。Observable和Subject是两个“生产”实体，Observer和Subscriber是两个“消费”实体。RxJava本质上是Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流，用Schedulers参数化异步数据流的并发处理，Rx可以这样定义：Rx = Observables + LINQ + Schedulers。

四、RxJava操作符的分类

操作符的分类
1.Creating ObServables(创建Observable)
2.Transforming Observables(转换Observabl）
3.Filtering Observables(过滤Observable)
4.Combing Observables(组合Observable)
5.Error Handling Operators(处理错误）

4.1Creating ObServables(创建Observable）

Create	Just	From
Defer	Empty/Never/Throw	Interval
Range	Repeat	Start
Timer	***	***

1.原始的创建：Observable.create()

private static void create() {  //一个原始的创建
  Observable.create(new OnSubscribe<String>(){

      @Override
      public void call(Subscriber<? super String> subscriber) {
          subscriber.onNext("RxJava学习");
      }
  }).subscribe(new Subscriber<String>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError()");
      }

      @Override
      public void onNext(String arg0) {
          System.out.println("onNext():"+arg0);
      }
      
  });
}

2.Observable.just() 方法可以传入一到九个参数，它们会按照传入的参数的顺序来发射它们。 just() 方法也可以接受列表或数组，

private static void just() {//对create的简写 快捷键
  Observable.just("sdgsagsag").subscribe(new Subscriber<String>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
          
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          
      }

      @Override
      public void onNext(String arg0) {
          System.out.println("onNext():"+arg0);
          
      }
  });
 }

3.Observable.from() 创建符可以从一个列表/数组来创建Observable,并一个接一个的从列表/数组中发射出来每一个对象，或者也可以从Java Future 类来创建Observable，并发射Future对象的 .get() 方法返回的结果值。传入 Future 作为参数时，我们可以指定一个超时的值。Observable将等待来自 Future 的结果；如果在超时之前仍然没有结果返回，Observable将会触发 onError() 方法通知观察者有错误发生了。

private static void from() {  //创建一个数组和列表转换成其他数据类型的对象 和数据类型
     ArrayList<Integer> items = new ArrayList<Integer>();
     items.add(1);
     items.add(2);
     items.add(3);
     items.add(5);
     items.add(6);
     items.add(7);
     Observable.from(items).subscribe(new Subscriber<Integer>(){

       @Override
       public void onCompleted() {
          System.out.println("onCompleted()");
            
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
            
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext():"+arg0);
            
      }     
    });
}

4.Observable.defer()对象后赋值也能输出来

private static String valuestr;
private static void defer() {//Observable对象后赋值 也能输出来
  ArrayList<Integer> items = new ArrayList<Integer>();
  items.add(1);
  items.add(2);
  items.add(3);
  items.add(4);
  items.add(5);
  items.add(6);
  items.add(7);
  Observable observable = Observable.defer(new Func0<Observable<String>>() {

      @Override
      public Observable<String> call() {
          return Observable.just(valuestr);
      }
  });
  
  valuestr = "速度更快sag凯撒";
  observable.subscribe(new Subscriber<String>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
          
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(String arg0) {
          System.out.println("onNext():"+arg0);
          
      }
      
  });

}

5.Obervable.range()对象后赋值也能输出来

private static void range() {//Observable对象后赋值 也能输出来
  ArrayList<Integer> items = new ArrayList<Integer>();
  items.add(1);
  items.add(2);
  items.add(3);
  items.add(4);
  items.add(5);
  items.add(6);
  items.add(7);
  Observable observable = Observable.range(1, 5).repeat(4);
  
  observable.subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
          
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext():"+arg0);
          
      }
    });
 }

4.2Transforming Observables(转换Observable)

1.Map:将一个对象转换为我们任何一个想要的对象
2.FlatMap:Observable.flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable 一个对象返回Observable对象一对多
3.GroupBy:通过一定的规则对数据列表进行分组和归类。
4.Buffer:一次性把Observable数据对象发射到我们的观察者

5.Scan:把前面的数据进行累加发射到观察者

private static void testStranfrom() { //整形转换成字符串
  Observable.just(123).map(new Func1<Integer, String>() {

      @Override
      public String call(Integer arg0) {
          // TODO Auto-generated method stub
          return arg0 +"";
      }
  }).subscribe(new Subscriber<String>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(String arg0) {
          System.out.println("onNext():"+arg0);
      }
      
  });
}

private static void textflatmap() { 
  Observable.just(1,2,3,4,5,6).flatMap(new Func1<Integer, Observable<? extends String>>() {

      @Override
      public Observable<? extends String> call(Integer arg0) {
          // TODO Auto-generated method stub
          return Observable.just(arg0+"");
      }
  }).subscribe(new Subscriber<String>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(String arg0) {
          System.out.println("onNext():"+arg0);
      }
      
  });
}

private static void groupby() { 
  Observable.just(1,2,3,4,5,6).groupBy(new Func1<Integer, Integer>() {

      @Override
      public Integer call(Integer arg0) {
          // TODO Auto-generated method stub
          return arg0 % 2; //分组的规则
      }
  }).subscribe(new Observer<GroupedObservable<Integer,Integer>>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted()");
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(final GroupedObservable<Integer, Integer> arg) {
          arg.subscribe(new Subscriber<Integer>(){

              @Override
              public void onCompleted() {
                  
              }

              @Override
              public void onError(Throwable arg0) {
                  
              }

              @Override
              public void onNext(Integer arg0) {
                  System.out.println("groupbt onNext():"+arg.getKey()+"data:"+arg0);
              }
              
          });
      }
  });
}

private static void buffer() { 
  Observable.range(1, 5).buffer(2).subscribe(new Observer<List<Integer>>(){

      @Override
      public void onCompleted() {
          
      }

      @Override
      public void onError(Throwable arg0) {
          
      }

      @Override
      public void onNext(List<Integer> arg0) {
          System.out.println("onNext:"+arg0);
      }
      
  });
}

private static void scan() { //每次据取两个数据项，然后向观察者发射
  Observable.range(1, 5).scan(new Func2<Integer,Integer,Integer>(){
      @Override
      public Integer call(Integer sum, Integer arg1) {
          // TODO Auto-generated method stub
          return sum + arg1;
      }
  }).subscribe(new Observer<Integer>(){

      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generatd method stube
          System.out.println("onNext:"+arg0);
      }
      
  });
}

4.3Filtering Observables(过滤Observable)

1.Debounce: 一定的时间内数据项后面没有产生数据将该数据发送出去
2.Distinct:去掉重复的
3.ElementAt:指定位置的数据
4.Filter:按照我们指定的规则来进行过滤得道我们想要的数据
5.First:取列表中的第一位数据
6.IgnoreElements: 忽略列表上的所有元素事件结束的时候回调Complement函数
7.Last:我们指定的最后数位项数据
8.Sample:对数据样本数据的采集，采集后发射到观察者定时的取样取到的数据放到一个列表中再发射。
9.Skip:跳跃指定数据项的数据然后取后面的数据
10.SkipLast:跳过数据的后几项
11.Take:取值的意思在数据列表中如果只取数据项的前几项

12.TakeLast:取数据列表的最后几项数据指定的几位数据

private static void testDebounce() { //在一定的时间内 数据项后面没有产生数据 将该数据发送出去
  Observable.create(new OnSubscribe<Integer>() {

      @Override
      public void call(Subscriber<? super Integer> arg0) {
          for(int i = 0; i <10; i++) {
              try {
                  Thread.sleep(1000);
                  arg0.onNext(1);
              } catch (InterruptedException e) {
                  arg0.onError(e);
              }
          }
      }
      
  }).debounce(1,TimeUnit.SECONDS).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          System.out.println("onCompleted():" + "\n");
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          System.out.println("onError:" + arg0+  "\n");
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generated method stub
          System.out.println("onNext:" + arg0+  "\n");
      }
      
  });
}

 private static void testDistinct() { //去掉重复的
  Observable.just(1, 2, 3, 3, 2).distinct().subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          System.out.println("onCompleted():" + "\n");
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          System.out.println("onError:" + arg0+  "\n");
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generated method stub
          System.out.println("onNext:" + arg0+  "\n");
      }
      
  });
｝

private static void testElementAt() { //指定位置的数据
  Observable.just(1, 2, 2, 4).elementAt(2).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted():" + "\n");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError:" + arg0+  "\n");
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext:" + arg0+  "\n");
      }
  });
}
private static void testFilter() { //去掉重复的 过滤
  Observable.just(1, 2, 3, 3, 2).distinct().filter(new Func1<Integer,Boolean>(){
      
      public Boolean call(Integer arg0) {
          return arg0 > 2;
      }
      
  }).subscribe(new Subscriber<Integer>(){
      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          System.out.println("onCompleted():" + "\n");
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          System.out.println("onError:" + arg0+  "\n");
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generated method stub
          System.out.println("onNext:" + arg0+  "\n");
      }
      
  });
}
private static void testFirst() { //指定位置的数据
  Observable.just(1, 2, 2, 4).first().subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted():" + "\n");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError:" + arg0+  "\n");
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext:" + arg0+  "\n");
      }
  });
}

private static void testIgnoreElement() {
  Observable.create(new OnSubscribe<Integer>(){

      @Override
      public void call(Subscriber<? super Integer> arg0) {
          // TODO Auto-generated method stub
          arg0.onNext(233);
          //throw new NullPointerException();
          
      }
      
  }).ignoreElements().subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          System.out.println("onCompleted():" + "\n");
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          System.out.println("onError:" + arg0+  "\n");
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generated method stub
          System.out.println("onNext:" + arg0+  "\n");
      }
      
  });
}

private static void testSample() { //定时对数据进行采样  再发送出去
  Observable.create(new OnSubscribe<Integer>() {

      @Override
      public void call(Subscriber<? super Integer> arg0) {
          for(int i = 0; i <10; i++) {
              try {
                  Thread.sleep(1000);
                  arg0.onNext(i);
              } catch (InterruptedException e) {
                  arg0.onError(e);
              }
          }
          arg0.onCompleted();
      }
      
  }).sample(1,TimeUnit.SECONDS).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          // TODO Auto-generated method stub
          System.out.println("onCompleted():" + "\n");
      }

      @Override
      public void onError(Throwable arg0) {
          // TODO Auto-generated method stub
          System.out.println("onError:" + arg0+  "\n");
      }

      @Override
      public void onNext(Integer arg0) {
          // TODO Auto-generated method stub
          System.out.println("onNext:" + arg0+  "\n");
      }
      
  });
}
  private static void testSkip() { //跳过多少项目
  Observable.just(1, 2, 2, 4).skip(3).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted():" + "\n");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError:" + arg0+  "\n");
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext:" + arg0+  "\n");
      }
  });
}

private static void testTake() { //取出多少项目
  Observable.just(1, 2, 2, 4).take(3).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted():" + "\n");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError:" + arg0+  "\n");
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext:" + arg0+  "\n");
      }
  });
}

4.3Combing Observables(组合Observable)

1.Zip:在一段时间内将最近的元素进行组合在一起
2.Merge: 按照时间先后顺序进行组合
3.StartWith: 在当前的数据之前插入我们想要的数据（数据类型或者Observable）
4.CombineLatest: 最新接触组合数据
5.Join:

6.SwithchOnNext:后面的覆盖前面的数据组合

  /**
 * 用来合并两个Observable发射的数据项，根据Func2函数生成一个新的值并发射出去。其中一个Observable发送数据结束或者出现异常后，
 * 另外一个Observable也将停在发射数据
 */
private static void testZip() {
  Observable<Integer> observable1 = Observable.just(10, 20, 30);
  Observable<Integer> observable2 = Observable.just(4, 8, 12, 16);
  Observable.zip(observable1, observable2, new Func2<Integer,Integer,Integer>(){

      @Override
      public Integer call(Integer arg0, Integer arg1) {
          return arg0 + arg1;
      }
  }).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted");
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError():"+arg0.getMessage());
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext arg0 "+arg0);
      }
  });
}

 /**
   * 将两个Observable发射的事件序列组合合并成一个事件序列，就像是Observable发射的一样。你可以简单的将它理解成两个Observable合并成一个Observable对象
   * 
   */
private static void testMerg() {
  Observable<Integer> observable1 = Observable.just(10, 20, 30);
  Observable<Integer> observable2 = Observable.just(4, 8, 12, 16);
  Observable.merge(observable1,observable2).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError():"+arg0.getMessage());
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext arg0 "+arg0);
          
      }
  });
}

 /**
   * 用于在源Observable发射的数据前插入数据.使用startWith(Interable<T>)我们还可以在源Observable发射的数据前插入Iterable
  */
private static void testStartWith() {
  Observable<Integer> first = Observable.just(1, 3, 5);
  Observable<Integer> second = Observable.just(2, 4, 6);
  first.startWith(second).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted");
          
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError():"+arg0.getMessage());
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext arg0 "+arg0);
          
      }
  });
}

/**
  * 用于将两个Observable最近发射的数据已经Func2函数的规则进展集合  5+2 5+4 5+6
  */
private static void testCombineLatest() {
  Observable<Integer> first = Observable.just(1, 3, 5);
  Observable<Integer> second = Observable.just(2, 4, 6);
  first.combineLatest(first, second, new Func2<Integer,Integer,Integer>(){

      @Override
      public Integer call(Integer arg0, Integer arg1) {
          // TODO Auto-generated method stub
          return arg0 + arg1;
      }
  }).subscribe(new Subscriber<Integer>(){

      @Override
      public void onCompleted() {
          System.out.println("onCompleted");
      }

      @Override
      public void onError(Throwable arg0) {
          System.out.println("onError():"+arg0.getMessage());
          
      }

      @Override
      public void onNext(Integer arg0) {
          System.out.println("onNext arg0 "+arg0);
      }
  });
}

/**
  * join()函数基于时间窗口将两个Observables发射的数据结合在一起，每个Observable在自己的时间窗口内独有有效的，都可以拿来组合
  */
private static void testJoin() {
  Observable<Integer> first = Observable.just(1, 3, 5);
  Observable<Integer> second = Observable.create(new OnSubscribe<Integer>(){

      @Override
      public void call(Subscriber<? super Integer> arg0) {
          try{
              for(int i = 0; i < 5; i++) {
                  arg0.onNext(i);
              }
              arg0.onCompleted();
          }catch(Exception e){
              e.printStackTrace();
          }
      }
  });
}

4.4 Error Handling Operators(处理错误)

Catch:捕获异常处理
**onErrorReturn: **遇见异常正常结束数据流
onErrorResumeNext:当数据遇见异常，用正常数据代替错误数据项不可以捕获到异常信息
onExceptionResumeNext:当有异常的时候，可以用正确的数据流替换掉，并且和、可以捕获到异常信息

五、Schedulers(调度器)种类

调度器(Scheduler)是RxJava以一种极其简单的方式解决多线程问题机制。
Schedulers(调度器)种类：
.io() I/O操作
.computation()计算计算类型
.immediate() 这个调度器允许你立即在当前线程执行你指定的工作
.newThread() 建立一个新的线程它为指定任务启动一个新的线程
.trampoline() 按照顺序排队处理的调度器将会按序处理队列，并运行队列中每一个任务