RxJava

Github地址

https://github.com/ReactiveX/RxJava

https://github.com/ReactiveX/RxAndroid

RxJava想必开发的朋友都听过这个框架，而最近几年也是非常的火热，到目前位置已经 Version 3.x。而做为Android开发的小伙伴，应该是比较熟悉了，retrofit + RxJava做为项目的主框架，实在是太多了，刚好现在也在学习这一块的东西，作为一个笔记，记录一下，相互学习一下，写的不好的地方，请各位大佬见谅。

ReactiveX的历史

ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET，社区网站是 reactivex.io。

RxJava到底是什么

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.简单理解就是RxJava是响应式扩展的javavm实现：一个通过使用可观察序列组合异步和基于事件的程序的库。其实就是响应式编程和异步事件流编程，优点 简洁。

Rx模式

使用观察者模式

• 创建：Rx可以方便的创建事件流和数据流
• 组合：Rx使用查询式的操作符组合和变换数据流
• 监听：Rx可以订阅任何可观察的数据流并执行操作

观察者模式，简单来说， 就是被观察者与观察者之间存在着一对多的关系，一个被观察者可以被多个观察者依赖，当被观察者变化时，会通知到观察者；

简化代码

• 函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态
• 简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码
• 异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制
• 轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种并发问题

使用Observable的优势

Rx扩展了观察者模式用于支持数据和事件序列，添加了一些操作符，它让你可以声明式的组合这些序列，而无需关注底层的实现：如线程、同步、线程安全、并发数据结构和非阻塞IO。

下面说的主要是Android方面的Rxjava的简单使用和分析

引入依赖

// 依赖RxAndroid 2X 的依赖库
// 增加RxJava 2X 的依赖库
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'
implementation 'io.reactivex.rxjava2:rxjava:2.0.7'

RxJava观察者模式涉及到的几个类

1. Observable：被观察者
2. Observer：观察者
3. Event：被观察者通知观察者的事件
4. Subscribe：订阅
   
   

file

Observable

public abstract class Observable<T> {
        // 实现订阅
    public void subscribe(Observer<T> observer){
                // 通过将传进来observer包装成CreateEmitter，用于回调
    CreateEmitter<T> emitter = new CreateEmitter<T>(observer);
    // 订阅成功的方法
    observer.onSubscribe(emitter);
    // 回调发射器emitter
    subscribe(emitter);
}
    // 订阅成功，回调
    public abstract void subscribe(Emitter<T> emitter);
}

这个类一个是，订阅，一个是emitter发射。

观察者Observer

public abstract class Observer<T> {
  // 订阅开始
  public abstract void onSubscribe(Disposable d);
  // 拿到事件，用于传递数据
  public abstract void onNext(T t);
  // 错误事件
  public abstract void onError(Throwable e);
  // 完成事件
  public abstract void onComplete();
}

Observer类的方法都是回调,但是要处理Disposable，在onDestory的时候，要终止传递。

Disposable

public interface Disposable {
        //主动解除订阅
        void dispose();
        //查询是否解除订阅 true 代表 已经解除订阅
        boolean isDisposed();
}

这个类主要是处理内存泄漏的，rxjava虽然好用，但是总所周知，容易遭层内存泄漏。也就说在订阅了事件后没有及时取阅，导致在activity或者fragment销毁后仍然占用着内存，无法释放。而disposable便是这个订阅事件，可以用来取消订阅。
这里介绍我自己使用的两种方案：

• 使用CompositeDisposable

• 在oError和onComplete后调用disposable.dispose()
  new Observer<Integer>() {
  private Disposable mDisposable;

      @Override
      public void onSubscribe(Disposable d) {
          Log.d(TAG, "subscribe");
          mDisposable = d;
      }
  
      @Override
      public void onNext(Integer value) {
          Log.d(TAG, "onNext: " + value);
              mDisposable.dispose();
      }
  
      @Override
      public void onError(Throwable e) {
      }
  
      @Override
      public void onComplete() {
      }
  }
  

onError和onComplete不可以同时调用的原因：每次掉用过onError或onComplete其中一个方法后，就会掉用dispose()方法，此时订阅取消，自然也就不能掉用另一个方法了。

请求下载图片

        // 起点
    Observable.just(PATH)  // 内部会分发  PATH Stirng  // TODO 第二步

     // TODO 第三步
    .map(new Function<String, Bitmap>() {
        @Override
        public Bitmap apply(String s) throws Exception {
            URL url = new URL(PATH);
            HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
            httpURLConnection.setConnectTimeout(5000);
            int responseCode = httpURLConnection.getResponseCode(); // 才开始 request
            if (responseCode == HttpURLConnection.HTTP_OK) {
                InputStream inputStream = httpURLConnection.getInputStream();
                Bitmap bitmap = BitmapFactory.decodeStream(inputStream);
                return bitmap;
            }
            return null;
        }
    })
    .map(new Function<Bitmap, Bitmap>() {
        @Override
        public Bitmap apply(Bitmap bitmap) throws Exception {
            return bitmap;
        }
    })
    .subscribeOn(Schedulers.io())    // 给上面代码分配异步线程
    .observeOn(AndroidSchedulers.mainThread()) // 给下面代码分配主线程;
    //分配线程
     //.compose(rxud())

    // 订阅 起点 和 终点 订阅起来
    .subscribe(

            // 终点
            new Observer<Bitmap>() {

                // 订阅开始
                @Override
                public void onSubscribe(Disposable d) {
                    // 预备 开始 要分发
                    // TODO 第一步
                   
                }

                // TODO 第四步
                // 拿到事件
                @Override
                public void onNext(Bitmap bitmap) {
                
                }

                // 错误事件
                @Override
                public void onError(Throwable e) {

                }

                // TODO 第五步
                // 完成事件
                @Override
                public void onComplete() {
                    
                }
    });

这可以看到顺序整体式U型结构，从起点到最下面，再回过去执行。

装饰模式

Rxjava就是观察者模式+装饰者模式
上面简单说了下观察者模式，接下来看看装饰模式在RxJava中的应用
简单画个图看看

file


简单理解这个装饰模式就是封包和拆包 ，下图是来源于网络
file


再来看下时序图


file


file


这两个时序图涵盖了整个过程，包括了其中的一些源码的分析，图片来源与网络。
最后简单来说下RxJava的操作符，操作符有很多个，记不住的话，在使用的时候，再去查查API


file

RxJava的原理和模式比较复杂，这里只是简单分享了一个皮毛，真正的还需要自己去看看源码。

王子猪 祝您天天开心。