Rxjava学习笔记

前言：rxjava是近几年来比较火的一个框架，是一个基于事件流的一个链式框架，可以很方便的进行线程切换，使用各种不同的简单的操作符将以前传统复杂的代码逻辑简单化，可能这么说比较笼统，举几个例子：

1.用户需要先注册账号，注册成功之后，需要做登录操作
2.app首页的数据列表需要两个甚至多个接口的数据一起返回，需要将数据合并，再展示给用户
3.网络请求失败之后，需要重新发起请求
有工作经验的同学都知道这种需求实现起来并不复杂，但是代码体现上要不就是嵌套过多，要不就是需要一大堆的逻辑控制代码，不太美观，但是使用rxjava的链式调用以及操作符（比如concat）就很简单了，并且比较美观

rxjava的学习曲线比较陡峭，一直想学习提升一下b格，无奈太懒，闲来无事看了看菜鸟窝的免费课程，学习了下，这里主要记录一下用法以及操作符，加深一下印象

注意：此笔记使用的版本为 rxjava3.x

Observable，subscribe，Observer

这三者为rxjava的基础，框架使用的是观察者模式，既然是观察者模式，那就存在订阅（subscribe）关系，逻辑上是“订阅者”订阅“被订阅者”，但在rxjava的代码体现上为“被订阅者”订阅“订阅者”。

Observable：被观察者，被订阅者
Observer：观察者，订阅者
subscribe：订阅

1.Observable

既然是被观察者订阅观察者，先说被观察者，被观察者的创建有很多种方式，常用的如下：

Observable observable = ObservableAll.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {
            }
});

被观察者为事件的发出方，rxjava中所有的事件操作符api调用之后，都会生成一个Observable对象；
注：操作符是用来定义事件发出的规律，规则，但是不论规则是怎么样的，最终都会生成一个Observable对象（个人理解，有问题的话请指正）

2.subscribe

subscribe即为订阅的动作，代码体现为：

//getObserver（）只是一个获取observer的方法，下面有说
observeble.subscribe(getObserver());

3.Observer

Observer为事件的观察者，即事件最终会通知到“观察者”，观察者的体现形式比较多：

//这里的String为事件最终通知的数据类型，可以为基本数据类型也可以为包装数据类型，bean类
new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {}

            @Override
            public void onNext(@NonNull String o) {}

            @Override
            public void onError(@NonNull Throwable e) {}

            @Override
            public void onComplete() {}
        };

以上这些东西合起来就组成了一个订阅关系，代码如下：

ObservableAll.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                           //emitter.onNext("1"); 该方法调用之后，对应的Observer的onNext会回调
                           //emitter.onNext("2");
                           //emitter.onComplete();该方法调用之后，对应的Observer的onComplete会回调
                           //如果发生异常，会调用Observer的onError方法；
         }
        }) .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) { }

                    @Override
                    public void onNext(@NonNull String s) {}

                    @Override
                    public void onError(@NonNull Throwable e) {}

                    @Override
                    public void onComplete() {}
                });

当然，这是rxjava的最基本使用

4.操作符

操作符是rxjava中比较重要的一环，可以用来定义事件的发出规则，比如：

1).ObservableAll.just(1, 2, 3, 4)：将参数1,2,3,4按顺序发出

2).ObservableAll.create（） ： 创建操作符，需要传入一个接口实现类，实现 subscribe 方法，该方法的参数emitter为事件的发射器，可以在此方法内定义数据源的操作或者使用发射器发射事件；个人认为该实现方式为最基础的创建Observable的方法，本身没有任何的规则或者规律；

ObservableAll.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                           //emitter.onNext("1"); 该方法调用之后，对应的Observer的onNext会回调
                           //emitter.onNext("2");
                           //emitter.onComplete();该方法调用之后，对应的Observer的onComplete会回调
                           //如果发生异常，会调用Observer的onError方法；
          }
        })

3).ObservableAll.fromarray(1, 2, 3, 4)

将参数1,2,3,4按顺序发出，作用同just操作符

4).ObservableAll.fromCallable（）

需要传入一个接口实现类Callable，实现 call 方法

ObservableAll.fromCallable(new Callable<String>() {
            @Override
            public String call() throws Exception {
                //有点类似于create操作符，但是在call内无法主动控制onSubscribe，onNext，onComplete
                return "";
            }
        });

5).ObservableAll.interval(0, 1, TimeUnit.SECONDS)

以某一个时间为间隔重复执行，第一个参数为延时执行时间，第二个参数为间隔时间，第三个参数为“秒”，也可以传“分”“时”；

6).merge

作用如同字面意思，需要传入多个Observable对象作为事件源，然后将这些事件合并发出，注意：该操作符是无序的，不会按传入的参数顺序来

7).concat

作用同merge，但是为有序，按传入的参数顺序来；

8).map

注意，该操作符需要先有数据源之后，才可以调用，非static，以上几个操作符都是确定数据源，返回一个Observerble对象，而该操作符为对数据源进行变换操作，代码如下：

//map操作符，将源数据变换之后依次发出
        ObservableAll.just(1, 2, 3, 4, 5).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Throwable {
                return integer.toString();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                Log.e("MainActivity", "map-----------------" + s);
            }
        });

这里是做了将数据源转为String的变换，当然可以做别的操作；

9).filter

该操作符的使用条件同上，作用为过滤，代码如下：

        ObservableAll.just(1, 2, 3)
                //filter为过滤操作符，对源数据进行过滤
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer >= 2;
                    }
                })

10).flatMap

将数据处理之后，返回一个Observebale<T>包装的数据

 ObservableAll.just(1, 2, 3, 4).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Throwable {
                return ObservableAll.just(integer + "");
            }
        });

暂时先写这么多，后续有用到的再补上