RxJava2.0学习总结

好吧。在不知道该做什么的时候，就跟随这别人的脚步走吧；
学习资料 ：
给初学者的RxJava2.0教程
强烈推荐这一系列教程，简单易懂，分析得面面俱到，虽然，我特么还是不会实战。
废物！
讲道理，我这保守落后的大清少年，在这之前我连这玩意是干什么的都不知道，现在也是半懵逼状态 。

在使用RxJava2.0之前

    compile 'io.reactivex.rxjava2:rxjava:2.0.1'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

说有个观察者，被观察者，他俩通过订阅来进行连接
Observer、Observable 、subscribe()

观察者通过这个订阅方法，告诉被观察者，你要做事情时啊，你就发送信息告诉我一声，
然后就是这样，被观察者呢，他就在自己进行操作的时候，发送信息给观察者，我要干事情喽。

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {

                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
//onComplete之后的数据，被观察者照常发送，但是观察者并不会接收到。
                e.onNext(4);
            }

        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i("TAG-->", "subscribe");
            }

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

            @Override
            public void onError(Throwable e) {

                Log.i("TAG-->", "error");
            }

            @Override
            public void onComplete() {
                Log.i("TAG-->", "complete");
            }
        });

还有RxJava可以很轻松的切换线程。

• Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
• Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
• Schedulers.newThread() 代表一个常规的新线程
• AndroidSchedulers.mainThread() 代表Android的主线程

   //RxJava内置的线程调度器，可以轻松的切换线程；

        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {

                Log.i("TAG-->", "Observable  thread  is :" + Thread.currentThread().getName());
                e.onNext(1);

            }
        });


        Consumer consumer = new Consumer<Integer>() {

            @Override
            public void accept(Integer integer) throws Exception {

                Log.d("TAG-->", "Observer thread is :" + Thread.currentThread().getName());
            }
        };

        //  observable.subscribe(consumer);

        observable.subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);

通过map转换数据类型

 //Map , 我们在上游发送integer类型数据，通过map ，将数据转换为 string  ,
        //  使得下游接受的数据为string

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {

            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This  is  result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {

                Log.i("TAG-->", s);
            }
        });

zip函数整合数据

 //打包数据 ，通过一个函数，将多个 Observable 发送的事件结合到一起，然后发送这些组合到一起的事件
        
        Observable observable1 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {

                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io());


        Observable observable2 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {

                e.onNext("A");
                e.onNext("B");
                e.onNext("C");
                e.onComplete();

            }
        }).subscribeOn(Schedulers.io());


        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {

            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).subscribe(new Observer() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object value) {
                Log.i("TAG-->", value + "");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });