RxJava 之响应式编程

什么是响应式编程

百度的解释: 响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

响应式编程是使用异步数据流进行编程.

主要理解 "异步" 跟 "数据流" 这个概念 . 举例子来 当你点开手机APP 的时候进行APP初始化业务.首先完成SDK初始化，数据库初始化，登陆，之后跳转主界面, SDK初始化，数据库初始化等等这些初始化在同时在异步中进行. 形成一个河流一样到最后的登录. (这借鉴网上的解释的栗子.)

其中数据流的概念

我 的理解就是事物跟事物之间一种传递的关系.上游的业务在自己工作完成之后，就需要通知下游，通知下游的方式有很多种，其中最棒的的方式就是通过数据（事件）流。每一个业务完成后，都会有一条数据（一个事件）流向下游，下游的业务收到这条数据（这个事件），才会开始自己的工作。

异步的概念

在APP初始化中，我们能发现SDK初始化，数据库初始化，登陆接口这三个业务本身相互独立，应当在不同的线程环境中执行，以保证他们不会相互阻塞。而假如没有异步编程，我们可能只能在一个线程中顺序调用这三个相对耗时较多的业务，最终再去做页面跳转，这样做不仅没有忠实反映业务本来的关系，而且会让你的程序“反应”更慢.异步编程本身是有很多优点的，比如挖掘多核心CPU的能力，提高效率，降低延迟和阻塞等等. RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。

现在用代码来了解RxJava跟响应式编程.

要在Android中使用RxJava2, 先添加Gradle配置:

  //解释一下  implementation 跟compile 的区别
  //compile依赖的确实可以做到依赖传递，但是AS 3.0开始推荐使用implementation取代了compile，
//依赖传递失效了.  而 ##implement## 的意思是将该依赖隐藏在内部，而不对外部公开. 在 app //mudule 中//使用 implement 依赖的第三方库, 在其他 mudule 是无法调用的,## compile ##android //studio 3.0 版本后废弃该指令 改用 api 代替, api 完全等同于之前的 compile 指令,
// 也就是普通的依赖, //第三方库在 mudule 中依赖后其他 mudule 都可以使用该库.官方推荐在不影响的前提下优先使用 //implement 指令依赖.
    implementation 'io.reactivex.rxjava2:rxjava:2.1.14'
    implementation 'io.reactivex.rxjava2:rxandroid:2.0.2' 

发射一个事件, 后面接收一个事件, 就跟河流一样, 上游发射,到下游就接收. 这是圈连起来的结果.

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);  //发射第一个事件
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {  //事件的响应拿到结果
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("----->", "subscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.d("----->", ""+integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.d("----->", "异常");
            }

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

分开来写

//创建一个上游 Observable：
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
        //创建一个下游 Observer
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d("----->", "subscribe");
            }

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

            @Override
            public void onError(Throwable e) {
                Log.d("----->", "error");
            }

            @Override
            public void onComplete() {
                Log.d("----->", "complete");
            }
        };
        //建立连接
        observable.subscribe(observer);

//打印的结果:
05-29 19:52:44.454 1435-1435/view.dome.com.rxjavadome D/----->: 
    subscribe
    1
    2
    3
    complete