RxJava常用操作符

• 创建

  • unfaseCreate(create)
    创建一个Observable（被观察者），当被观察者（Observer）/订阅者（Subscriber）
    subscribe(订阅)的时候就会依次发出三条字符串数据（"Hello","RxJava","Nice to meet you"）
    最终onComplete

Observable.unsafeCreate(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello");
                subscriber.onNext("RxJava");
                subscriber.onNext("Nice to meet you");
                subscriber.onCompleted();
            }
        });

• just
  作用同上，订阅时依次发出三条数据，不过此方法参数可以有1-9条

Observable.just("Hello", "RxJava", "Nice to meet you")

• from
  作用同just不过是把参数封装成数组或者可迭代的集合在依次发送出来，突破了just9个参数的限制

String[] strings = {"Hello", "RxJava", "Nice to meet you"};
Observable.from(strings)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

• defer
  顾名思义，延迟创建

private String[] strings1 = {"Hello", "World"};
    private String[] strings2 = {"Hello", "RxJava"};

    private void test() {
        Observable<String> observable = Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.from(strings1);
            }
        });

        strings1 = strings2; //订阅前把strings给改了
        observable.subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

我们发现数据结果变成这样了

onNext--> Hello
onNext--> RxJava
onComplete

由此可以证明defer操作符起到的不过是一个“预创建”的作用，真正创建是发生在订阅的时候

• empty
  创建一个空的，不会发射任何事件（数据）的Observable

Observable.empty()
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onComplete

• never
  创建一个不会发出任何事件也不会结束的Observable

Observable.never()
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

······

• error
  创建一个会发出一个error事件的Observable

Observable.error(new RuntimeException("fuck!"))
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onError--> fuck!

• range
  创建一个发射一组整数序列的Observable

Observable.range(3, 8)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
onNext--> 7
onNext--> 8
onNext--> 9
onNext--> 10
onComplete

• interval
  创建一个无限的计时序列，每隔一段时间发射一个数字（从0开始）的Observable

   Observable.interval(1, TimeUnit.SECONDS)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

   System.in.read();//阻塞当前线程，防止JVM结束程序

onNext--> 0
onNext--> 1
onNext--> 2
onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
...

• 转换

  • buffer(int count)
    将原发射出来的数据已count为单元打包之后在分别发射出来

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .buffer(3)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> [1, 2, 3]
onNext--> [4, 5, 6]
onNext--> [7, 8, 9]
onNext--> [10]
onComplete

• map
  将原Observable发射出来的数据转换为另外一种类型的数据

Observable.just("Hello", "RxJava", "Nice to meet you")
        .map(new Func1<String, Integer>() { //泛型第一个类型是原数据类型，第二个类型是想要变换的数据类型
            @Override
            public Integer call(String s) {
                // 这是转换成了Student类型
                // Student student = new Student();
                // student.setName(s);
                // return student;
                return s.hashCode();        //将数据转换为了int（取得其hashCode值）
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 69609650
onNext--> -1834252888
onNext--> -1230747480
onComplete

• flatMap
  作用类似于map又比map强大，map是单纯的数据类型的转换，而flapMap可以将原数据新的Observables，再将这些Observables的数据顺序缓存到一个新的队列中，在统一发射出来

Observable.just("Hello", "RxJava", "Nice to meet you")
        .flatMap(new Func1<String, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(String s) {
                return Observable.just(s.hashCode());
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> 69609650
onNext--> -1834252888
onNext--> -1230747480
onComplete

这里虽然结果和map是相同的，但是过程却是不同的。flatMap是先将原来的三个字符串（"Hello","RxJava","Nice to meet you"）依次取其hashCode，在利用Observable.just将转换之后的int类型的值在发射出来。map只是单穿的转换了数据类型，而flapMap是转换成了新的Observable了，这在开发过程中遇到嵌套网络请求的时候十分方便。

• window
  看到网上其他人说他的总用类似于buffer，不过我倒是认为他更像flatMap，区别的是flapMap在转换之后形成新的Observable会再将新的数据发射出来，不过window就仅仅只转换成了发射新的数据类型的Observable，有点像是flatMap在干活时半途而废的意思。

• 过滤

  • filter
    对发射的数据做一个限制，只有满足条件的数据才会被发射

Observable.just("Hello", "RxJava", "Nice to meet you")
        .filter(new Func1<String, Boolean>() {
            @Override
            public Boolean call(String s) {
                //这里的显示条件是s的长度大于5，而Hello的长度刚好是5
                //所以不能满足条件
                return s.length() > 5;
            }
        })
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> RxJava
onNext--> Nice to meet you
onComplete

• take
  只发射前N项的数据（takeLast与take想反，只取最后N项数据）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .take(2)
        //.taktLast(2)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> Hello
onNext--> RxJava
onComplete

• skip
  发射数据时忽略前N项数据（skpiLast忽略后N项数据）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .skip(2)
        //.skipLast(2)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
       }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> Nice to meet you
onComplete

• elementAt
  获取原数据的第N项数据作为唯一的数据发射出去（elementAtOrDefault会在index超出范围时，给出一个默认值发射出来）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .elementAtOrDefault(1, "Great")
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> RxJava
onComplete

• distinct
  过滤掉重复项

Observable.just("Hello", "Hello", "Hello", "RxJava", "Nice to meet you")
        .distinct()
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> Hello
onNext--> RxJava
onNext--> Nice to meet you
onComplete

• 组合

  • startWith
    在发射一组数据之前先发射另一组数据

Observable.just("Hello", "RxJava", "Nice to meet you")
        .startWith("One", "Two", "Three")
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> One
onNext--> Two
onNext--> Three
onNext--> Hello
onNext--> RxJava
onNext--> Nice to meet you
onComplete

• merge
  将多个Observables发射的数据合并后在发射

Observable.merge(Observable.just(1, 2, 3), Observable.just(4, 5),
        Observable.just(6, 7), Observable.just(8, 9, 10))
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 1
onNext--> 2
onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
onNext--> 7
onNext--> 8
onNext--> 9
onNext--> 10
onComplete

• zip
  按照自己的规则发射与发射数据项最少的相同的数据

Observable.zip(Observable.just(1, 8, 7), Observable.just(2, 5),
        Observable.just(3, 6), Observable.just(4, 9, 0), new Func4<Integer, Integer, Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2, Integer integer3, Integer integer4) {
                return integer < integer2 ? integer3 : integer4;
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 3
onNext--> 9
onComplete

通过观察以上例子可以发现我们的发射规则是如果发射的第一个数据小于第二个数则发射第三个数据，否则发射第四个数据（我们来验证一下，1确实是小于2的，随意发射的是3；8并不小于5，所以发射的是9，又因为四个发射箱，最少的之后两项，所以最后只发射了两项数据）