RxJava操作符小结

本篇基于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项数据。

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 219,589评论 6赞 508
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 93,615评论 3赞 396
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 165,933评论 0赞 356
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 58,976评论 1赞 295
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 67,999评论 6赞 393
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 51,775评论 1赞 307
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,474评论 3赞 420
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,359评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,854评论 1赞 317
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,007评论 3赞 338
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,146评论 1赞 351
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,826评论 5赞 346
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,484评论 3赞 331
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,029评论 0赞 22
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,153评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,420评论 3赞 373
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,107评论 2赞 356

RxJava操作符小结

1、Single

2、Subject

3、操作符

推荐阅读更多精彩内容