RxJava 操作符（过滤、组合）

过滤操作符

过滤操作符是过滤和选择Observable发射的数据序列，让Observable只返回满足条件的数据。

• filter
  filter 操作符是对原Observable产生的结果自定义规划进行过滤，只有满足条件才会提交给订阅者

       Observable.just(1, 2, 3, 5)
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        // 限制条件
                        return integer > 2;
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " filter =" + integer);
                    }
                });

E/zpan:  filter =3
E/zpan:  filter =5

• elementAt
  elementAt 操作符用来返回指定位置的数据。
  和它类似的有elementAtOrDefault(int,T),它可以有默认值

      Observable.just(1, 2, 3, 4)
                .elementAt(3)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " element =" + integer);
                    }
                });

E/zpan:  element =4

• distinct
  distinct 操作符用来去重，其只允许还没有发射过的数据项通过。
  和它类似的还有 distinctUntilChanged操作符，其用来去掉连续重复的数据。

        Observable.just(1, 4, 8, 5, 5, 1, 5, 6, 8)
                .distinct()
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " distinct=" + integer);
                    }
                });

E/zpan:  distinct=1
     distinct=4
     distinct=8
     distinct=5
     distinct=6

• skip
  skip 操作符将原Observable发射的数据过滤掉前n项。
• skipLast
  skipLast 操作符将原Observable发射的数据从后面进行过滤操作

        Observable.just(1, 2, 3, 4, 5)
                .skip(2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " skip = " + integer);
                    }
                });

E/zpan:  skip = 3
     skip = 4
     skip = 5

• take
  take 操作符将原Observable发射的数据只取前n项
• takeLast
  takeLast 操作符将原Observable发射的数据从后面进行过滤操作

        Observable.just(1, 2, 3, 4, 5)
                .take(2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " take = " + integer);
                    }
                });

E/zpan:  take = 1
     take = 2

• ignoreElements
  ignoreElements 操作符忽略所有原Observable产生的结果，只把Observable的OnCompleted和onError事件通知给订阅者

        Observable.just(1, 2, 3, 4)
                .ignoreElements()
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.e("zpan", "==onCompleted=====");
                    }

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

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("zpan", "==onNext=====");
                    }
                });

E/zpan: ==onCompleted=====

• ofType
  过滤指定类型的数据

        Observable.just(1, 2, "3", "4")
                .ofType(String.class)
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.e("zpan", "ofType = " + s);
                    }
                });

E/zpan: ofType = 3
    ofType = 4

• first/firstOrDefault
  只发射满足条件的第一项，后者可以设置默认值。
• last/lastOrDefault
  只发射满足条件的最后一项，后者可以设置默认值。

        Observable.just(1, 2, 3, 4)
                .first(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer > 2;
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan"," first =" + integer);
                    }
                });

E/zpan:  first =3

• throttleFirst
  throttleFirst 操作符会定期发射这个时间段里原Observable发射的第一个数据，throttleFirst 操作符默认在computation调度器上执行。与throttleFirst类似的有sample操作符，它会定时的发射源Observable最近发射的数据，其他都会被过滤掉。

       Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 10; i++) {
                    subscriber.onNext(i);
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    subscriber.onCompleted();
                }
            }
        })
                // 每隔100ms发射一个数据，throttleFirst设定的时间是200ms，所以它会发射每个200ms内的第一个数据
                .throttleFirst(200, TimeUnit.MILLISECONDS)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " throttleFirst=" + integer);
                    }
                });

组合操作符

组合操作符可以同时处理多个Observable来创建我们所需要的Observable。

• startWith
  startWith 操作符会在原Observable发射的数据前面插上一些数据。

        Observable.just(4,5,6)
                .startWith(1,2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan"," startWith = " + integer);
                    }
                });

E/zpan:  startWith = 1
     startWith = 2
     startWith = 4
     startWith = 5
     startWith = 6

• merge
  merge 操作符将多个Observable合并到一个Observable中进行发射，merge可能会让合并的Observable发射的数据交错。

Observable<Integer> o1 = Observable.just(1, 2, 3);
        Observable<String> o2 = Observable.just("7", "8", "9");
        Observable.merge(o1,o2)
                .subscribe(new Action1<Serializable>() {
                    @Override
                    public void call(Serializable serializable) {
                        Log.e("zpan","merge = " + serializable.toString());     
                    }
                });

E/zpan: merge = 1
    merge = 2
    merge = 3
    merge = 7
    merge = 8
    merge = 9

• concat
  类似于merge，只是concat严格按照顺序发射数据。

        Observable<Integer> o1 = Observable.just(1, 2);
        Observable<Integer> o2 = Observable.just(6, 7);
        Observable.concat(o1, o2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e("zpan", " concat = " + integer);
                    }
                });

E/zpan:  concat = 1
     concat = 2
     concat = 6
     concat = 7

• zip
  zip操作符合并两个或多个Observable发射出的数据，根据指定的函数变换它们，并发射一个新的值。

        Observable<Integer> o1 = Observable.just(1, 2);
        Observable<String> o2 = Observable.just("- A", "- B");
        Observable.zip(o1, o2, new Func2<Integer, String, Object>() {
            @Override
            public Object call(Integer integer, String s) {
                return integer + s;
            }
        })
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {
                        Log.e("zpan", " zip = " + o);
                    }
                });

E/zpan:  zip = 1- A
E/zpan:  zip = 2- B

• CombineLatest
  combineLatest 操作符和zip操作符类似。
  将两个Observale最近发射的数据已经Func2函数的规则进展组合。

        Observable<Integer> o1 = Observable.just(1, 2);
        Observable<String> o2 = Observable.just("- A", "- B");
        Observable.combineLatest(o1, o2, new Func2<Integer, String, Object>() {
            @Override
            public Object call(Integer integer, String s) {
                return integer + s;
            }
        })
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {
                        Log.e("zpan", " combineLatest= " + o);
                    }
                });

E/zpan:  combineLatest = 2- A
E/zpan:  combineLatest = 2- B

• join
  结构：
  onservableA.join(observableB, 控制observableA发射数据有效期的函数， 控制observableB发射数据有效期的函数，两个observable发射数据的合并规则)
  join操作符的效果类似于排列组合，把第一个数据源A作为基座窗口，他根据自己的节奏不断发射数据元素，第二个数据源B，每发射一个数据，我们都把它和第一个数据源A中已经发射的数据进行一对一匹配；

        Observable<Integer> o1 = Observable.just(1, 2);
        Observable<String> o2 = Observable.just("- c", "- d");
        o1.join(o2, new Func1<Integer, Observable<Long>>() {
            @Override
            public Observable<Long> call(Integer integer) {
                return Observable.timer(1, TimeUnit.SECONDS);
            }
        }, new Func1<String, Observable<Long>>() {
            @Override
            public Observable<Long> call(String s) {
                return Observable.timer(0, TimeUnit.SECONDS);
            }
        }, new Func2<Integer, String, Object>() {
            @Override
            public Object call(Integer integer, String s) {
                return integer + s;
            }
        })
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {
                        Log.e("zpan", " join= " + o);
                    }
                });

E/zpan:  join= 2- c
     join= 1- c
     join= 2- d
     join= 1- d