RxJava操作符小结

2019-03-19 本文已影响0人 Dane_404

本篇基于RxJava中文文档https://mcxiaoke.gitbooks.io/rxdocs/content/做个小结，温故知新。

1、Single

Single类似于Observable，不同的是，它总是只发射一个值，或者一个错误通知，而不是发射一系列的值。
订阅Single只需要两个方法：
onSuccess：Single发射单个值到这个方法。
onError：如果无法发射需要的值，Single发射一个Throwable对象到这个方法。
Single只会调用这两个方法中的一个，而且只会调用一次，调用了任何一个方法之后，订阅关系终止。

2、Subject

Subject可以看成一个桥梁或代理，它同时充当了Observer和Observable的角色。因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。
由于一个Subject订阅一个Observable，它可以触发这个Observable开始发射数据（如果那个Observable是"冷"的--就是说，它等待有订阅才开始发射数据）。因此有这样的效果，Subject可以把原来那个"冷"的Observable变成"热"的。

AsyncSubject
一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。（如果原始Observable没有发射任何值，AsyncObject也不发射任何值）它会把这最后一个值发射给任何后续的观察者。

  AsyncSubject<Integer> subject = AsyncSubject.<Integer>create();
  Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(4);
      }
  }).subscribe(subject);
  subject.onComplete();  //要调onComplete()，下游才会收到
  subject.subscribe(new Observer<Integer>() {
      @Override
      public void onSubscribe(Disposable d) {

      }

      @Override
      public void onNext(Integer value) {
          Log.d("-------", ""+value);  //收到4
      }

      @Override
      public void onError(Throwable e) {

      }

      @Override
      public void onComplete() {

      }
  });

BehaviorSubject
当观察者订阅BehaviorSubject时，它开始发射原始Observable最近发射的数据（如果此时还没有收到任何数据，它会发射一个默认值），然后继续发射其它任何来自原始Observable的数据。

  //BehaviorSubject.create()没默认值
  BehaviorSubject<Integer> subject = BehaviorSubject.<Integer>createDefault(1);  
  subscribe(subject);  //在这里订阅将收到1,1,2,4
  Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(4);
      }
  }).subscribe(subject);
  //subject.onComplete();   在订阅之前调用则收不到任何数据
 // subscribe(subject);  //在这里订阅将收到4
 private void subscribe(BehaviorSubject<Integer> subject) {
  subject.subscribe(new Observer<Integer>() {
      @Override
      public void onSubscribe(Disposable d) {

      }

      @Override
      public void onNext(Integer value) {
          Log.d("-------", ""+value); 
      }

      @Override
      public void onError(Throwable e) {

      }

      @Override
      public void onComplete() {

      }
  });
 }

PublishSubject
PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。

  PublishSubject<Integer> subject = PublishSubject.<Integer>create();
  subscribe(subject);  
  Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(4);
      }
  }).subscribe(subject);
  //subscribe(subject);  //这里将收不到

ReplaySubject
ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。也有其它版本的ReplaySubject，在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据（原始Observable发射的）。

3、操作符

Defer
直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable。

  Observable<Integer> objectObservable = Observable.defer(new Callable<ObservableSource<Integer>>() {
      @Override
      public ObservableSource<Integer> call() throws Exception {
          return Observable.just(a);  //此时a为0
      }
  });
  a = 1;
  objectObservable.subscribe(new Observer<Integer>() {
      @Override
      public void onSubscribe(Disposable d) {

      }

      @Override
      public void onNext(Integer value) {
          Log.d("-------",value+"");   //输出1
      }

      @Override
      public void onError(Throwable e) {

      }

      @Override
      public void onComplete() {

      }
  });

Empty、Never、Throw
Empty：创建一个不发射任何数据但是正常终止的Observable。
Never：创建一个不发射数据也不终止的Observable。
Throw：创建一个不发射数据以一个错误终止的Observable。

  Observable.empty().subscribe(new Observer<Object>() {
      @Override
      public void onSubscribe(Disposable d) {
          Log.d("----","onSubscribe");   //Never 只走onSubscribe
      }

      @Override
      public void onNext(Object value) {

      }

      @Override
      public void onError(Throwable e) {
          Log.d("----","onError");  //Throw只走，onError，onComplete
      }

      @Override
      public void onComplete() {
          Log.d("----","onComplete");  //Empty不会走onNext，只走，onSubscribe，onComplete
      }
  });

From
将其它种类的对象和数据类型转换为Observable

  Integer[] items = { 0, 1, 2, 3, 4, 5 };
  Observable.fromArray(items).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("--------",integer+""); //依次打印0到5
      }
  });
  Observable.fromFuture(new Future<Integer>() {
      @Override
      public boolean cancel(boolean mayInterruptIfRunning) {
          return false;
      }

      @Override
      public boolean isCancelled() {
          return false;
      }

      @Override
      public boolean isDone() {
          return false;
      }

      @Override
      public Integer get() throws InterruptedException, ExecutionException {
          return 1;
      }

      @Override
      public Integer get(long timeout, @NonNull TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
          return null;
      }
  }).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("--------",integer+""); //Future.get()方法返回的单个数据
      }
  });
  List<Integer> list = new ArrayList<>();  //集合继承了Iterable
  list.add(1);
  list.add(2);
  Observable.fromIterable((Iterable<Integer>) list).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("--------",integer+""); //依次收到1,2
      }
  });

Interval
创建一个按固定时间间隔发射整数序列的Observable。

  //参数一是间隔几秒开始发送，参数二是发送间隔
  mSubscribe = Observable.interval(4000, 2000, TimeUnit.MILLISECONDS).subscribe(new Consumer<Long>() {
      @Override
      public void accept(Long aLong) throws Exception {
          Log.d("----------", Thread.currentThread().getName());  //interval默认子线程
          Log.d("-------", aLong + "");
          if (aLong == 5.0){   
              mSubscribe.dispose();
          }
      }
  });

Just
创建一个发射指定值的Observable。

   Observable.just(1,2,3).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("---------",integer+"");
      }
  });

Range
发射一个范围内的有序整数序列，你可以指定范围的起始和长度。

  Observable.range(1,10).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("-------",integer+"");  //依次收到1到10
      }
  });

Repeat
创建一个发射特定数据重复多次的Observable。

  Observable.just(1).repeat(3).subscribe(new Consumer<Integer>() {
     @Override
     public void accept(Integer integer) throws Exception {
         Log.d("-------",integer+"");   //收到3次
     }
 });

Timer
创建一个Observable，它在一个给定的延迟后发射一个特殊的值。

  Observable.timer(1000, TimeUnit.MILLISECONDS).subscribe(new Consumer<Long>() {
      @Override
      public void accept(Long aLong) throws Exception {
          Log.d("------",aLong+"");  //它在延迟一段给定的时间后发射一个简单的数字0
      }
  });

Buffer
定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值。

  Observable.just(1,2,3,4,5,6).buffer(3).subscribe(new Consumer<List<Integer>>() {
      @Override
      public void accept(List<Integer> integers) throws Exception {
          Log.d("--------",integers+""); //输出[1,2,3]、[4,5,6]
      }
  });
  Observable.just(1,2,3,4,5,6).buffer(3,2).subscribe(new Consumer<List<Integer>>() {
      @Override
      public void accept(List<Integer> integers) throws Exception {
          Log.d("--------",integers+""); //输出[1,2,3]、[3,4,5]、[5,6]
      }
  });

Map
对Observable发射的每一项数据应用一个函数，执行变换操作

Observable.create(new ObservableOnSubscribe<Integer>() {
     @Override
     public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         e.onNext(1);
         e.onNext(2);
         e.onNext(3);
     }
 }).map(new Function<Integer, String>() {
     @Override
     public String apply(Integer integer) throws Exception {
         return "转换"+integer;
     }
 }).subscribe(new Consumer<String>() {
     @Override
     public void accept(String s) throws Exception {
         Log.d("--------",s);
     }

Cast
cast操作符将原始Observable发射的每一项数据都强制转换为一个指定的类型，然后再发射数据，它是map的一个特殊版本。

 Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).cast(String.class).subscribe(new Consumer<String>() {
      @Override
      public void accept(String s) throws Exception {
          Log.d("--------",s);
      }
  });

FlatMap
FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable。注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).flatMap(new Function<Integer, ObservableSource<String>>() {
      @Override
      public ObservableSource<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("-------------",s);  //并不一定就是1,2,3的顺序
      }
  });

FlatMapIterable
flatMapIterable这个变体成对的打包数据，然后生成Iterable而不是原始数据和生成的Observables，但是处理方式是相同的。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).flatMapIterable(new Function<Integer, Iterable<String>>() {
      @Override
      public Iterable<String> apply(Integer integer) throws Exception {
          List<String> list = new ArrayList<String>();
          list.add(integer+"---");
          return list;
      }
  }).subscribe(new Consumer<String>() {
      @Override
      public void accept(String s) throws Exception {
          Log.d("------",s);
      }
  });

concatMap
它和flatMap的作用几乎一模一样, 只是它的结果是严格按照上游发送的顺序来发送的。

switchMap
当原始Observable发射一个新的数据（Observable）时，它将取消订阅并停止监视产生执之前那个数据的Observable，只监视当前这一个。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).switchMap(new Function<Integer, ObservableSource<String>>() {
      @Override
      public ObservableSource<String> apply(Integer integer) throws Exception {

          return Observable.just(integer+"").subscribeOn(Schedulers.io());  //并发
      }
  }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
      @Override
      public void accept(String s) throws Exception {
          Log.d("------",s);   //只打印出3
      }

groupBy
GroupBy操作符将原始Observable分拆为一些Observables集合，它们中的每一个发射原始Observable数据序列的一个子序列。哪个数据项由哪一个Observable发射是由一个函数判定的，这个函数给每一项指定一个Key，Key相同的数据会被同一个Observable发射。
注意：groupBy将原始Observable分解为一个发射多个GroupedObservable的Observable，一旦有订阅，每个GroupedObservable就开始缓存数据。因此，如果你忽略这些GroupedObservable中的任何一个，这个缓存可能形成一个潜在的内存泄露。

 Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).groupBy(new Function<Integer, String>() {
      @Override
      public String apply(Integer integer) throws Exception {
          return integer+"";  //返回值相同的在同一组
      }
  }).subscribe(new Consumer<GroupedObservable<String, Integer>>() {
      @Override
      public void accept(final GroupedObservable<String, Integer> stringIntegerGroupedObservable) throws Exception {
          stringIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
              @Override
              public void accept(Integer integer) throws Exception {
                  Log.d("---------",stringIntegerGroupedObservable.getKey()+"---"+integer+"");
              }
          });
      }
  });

Scan
Scan操作符对原始Observable发射的第一项数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。它持续进行这个过程来产生剩余的数据序列。

 Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).scan(new BiFunction<Integer, Integer, Integer>() {
      @Override  //其中一个参数是接收的值，另一个是累加的值
      public Integer apply(Integer integer, Integer integer2) throws Exception {
          return integer*integer2;    //第一次是1，累加的值是0，输出1，第二次是2，累加的值是1,2*1=2
      }
  }).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("-------",integer+"");
      }
  });

window
Window和Buffer类似，但不是发射来自原始Observable的数据包，它发射的是Observables，这些Observables中的每一个都发射原始Observable数据的一个子集，最后发射一个onCompleted通知。

Distinct
抑制（过滤掉）重复的数据项。

 Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(1);
          e.onNext(3);
          e.onNext(2);
          e.onNext(3);
      }
  }).distinct().subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出1,3,2
      }
  });

Debounce
仅在过了一段指定的时间还没发射数据时才发射一个数据。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(1);
          e.onNext(3);
          e.onNext(2);
          e.onNext(3);
      }
  }).debounce(100, TimeUnit.MILLISECONDS).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出3
      }
  });

ElementAt
只发射第N项数据

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).elementAt(0).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出1
      }
  });

Filter
过滤操作

 Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).filter(new Predicate<Integer>() {
      @Override
      public boolean test(Integer integer) throws Exception {
          return integer == 2;
      }
  }).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出2
      }
  });

ofType
ofType是filter操作符的一个特殊形式。它过滤一个Observable只返回指定类型的数据。
First
只发射第一项（或者满足某个条件的第一项）数据。
IgnoreElements
不发射任何数据，只发射Observable的终止通知。
Last
只发射最后一项（或者满足某个条件的最后一项）数据。

Sample
定期发射Observable最近发射的数据项。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).sample(10, TimeUnit.MILLISECONDS).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出3
      }
  });

Skip
抑制Observable发射的前N项数据。

Observable.create(new ObservableOnSubscribe<Integer>() {
      @Override
      public void subscribe(ObservableEmitter<Integer> e) throws Exception {
          e.onNext(1);
          e.onNext(2);
          e.onNext(3);
      }
  }).skip(1).subscribe(new Consumer<Integer>() {
      @Override
      public void accept(Integer integer) throws Exception {
          Log.d("------",integer+"");     //输出2,3，忽略了1
      }
  });

SkipLast
抑制Observable发射的后N项数据。
Take
只发射前面的N项数据。
TakeLast
发射Observable发射的最后N项数据。

RxJava操作符小结

1、Single

2、Subject

3、操作符

猜你喜欢

热点阅读