Android框架第（六）篇---RxJava

**版权声明：本文为小斑马伟原创文章，转载请注明出处!

上一篇博客介绍了Retrofit使用。然而Retrofit使用同时，往往会伴随RxJava相结合。这样才会使网络请求流程更加清晰，线程控制更加简单。所以本篇先介绍下RxJava，在下一篇将会介绍如何使用Retrofit和RxJava相结合来完成一个网络请求。关于RxJava相关的原理和操作符介绍见---Android框架第（三）篇--响应式编程

一、 ReactiveX

ReactiveX是Reactive Extensiions的缩写，一般简写为Rx 就是一个函数库。Rx = Observables + LINQ + Schedulers （可发者利用可观察者事例和LINQ风格查询操作符来编写异步基于事件的程序）。开发者可以使用Observables来使用异步数据流，利用LINQ操作符来查询异步数据流，使用Schedulers 参数化异步数据流并发处理。

二、 ReactiveX原理

RxJava 是ReactiveX的一种java实现。一个在Java VM上使用可观测的序列来组成异步的、基于事件的程序库， 一个实现异步操作符的库，而且这个库是基于事件的。RxJava的异步操作是通过扩展的观察者模式实现Observable被观察者、Observer观察者、subscribe订event事件、Observable 和Observer通过subscribe()方法实现订阅关系就是创建一个被观察者对象来干活，然后使用各种操作符来是实现链式操作，就如同流水线一样，把我们想要处理的数据，一步一步处理成我们需要的产品。然后在发给我们观察者。

三、 ReactiveX引入

compile 'io.reactivex:rxjava:XXXX'
compile 'io.reactivex:rxandroid:XXXXX'

四、 ReactiveX使用步骤

• 1.创建Observer(观察者）决定事件触发的时候将有怎样的行为
• 2.创建Observable(被观察者)决定什么时候触发事件以及触发怎样的事件
• 3.Subscribe(订阅）将观察者和被观察者联结起来

五、创建Observer(观察者）

Observer<String> observer = new Observer<String>() {
    @Override
    public void onCompleted() { //事件队列完结 在RxJava中 不仅会把每个事件单独处理，还会把它们看成是一个队列。当没有新的事件发生时，这个时候就会触发oCompleted方法，这个方法也可以看成是事件完成标志。         Log.i(TAG,"onCompleted");
    }
    @Override
    public void onError(Throwable e) { //事件队列异常 在事件处理的过程中，难免会发生异常，在发生异常的时候，onError方法就会触发。同时事件队列就会终结
         Log.i(TAG,"onError");
    }
    @Override
    public void onNext(String s) { //就是一件将要处理的普通的事件，添加到处理事件队列中
         Log.i(TAG,"onNext"+s);
    }  
    @Override
    public void onStart() { //事件在还没发生前调用，这个方法用作一些做准备工作，比如数据的重置、清零
         Log.i(TAG,"onStart");
    }
};

Observer对象的抽象类Subscriber ,onSubscriber取消订阅 防止内存泄漏。

六、创建Observerable(被观察者）

Observable observable = Observable.create(new Observable.OnSubscribe<String>() { 
@Override
public void call(Subscriber<? super String> subsciber) {
     subscriber.onNext("hello");
     subscriber.onNext("hello");
     subscriber.onCompleted();  //实现了被观察者向观察者事件传递
     }
});

决定什么时候触发事件，以及触发怎么样的事件。使用create方法创建一个被观察者，同时为它定义一些事件触发的规则。传递一个OnSubscribe对象，这个对象会存储在Observable中，它的作用就相当于一个计划表，当我们的被观察者订阅的时候，OnSubscribe的call方法就会被调用。事件序列就会按照方法类依次触发。
RxJava用来快捷事件队列的方法

• 1、Observale.create() //基本的使用
• 2、Observable.just("hello","weiwei","niaho") 将传入的参数依次发送出来
• 3.、String[] words = { "Hello","weiwei","xiaobanma"};
  Observable observable = Observable.from(words); 将传入的数组拆分成具体的对象，然后依次的发送出来。

七、Subscribe(订阅）

observable.subscribe(observer);//或者observable.subscribe(subscriber);目的就是将观察者和被观察者关联起来。

八、Scheduler（线程控制器）

RxJava通过Scheduler来指定每一段代码应该运行在那个线程。不指定线程的情况，RxJava执行线程不变的原则，就在那个线程调用RxJava的方法就是那个线程生成事件，在那个线程生产事件，就会在那个线程销毁事件。所做的都是运行在主线程的，如果我们不指定线程，默认是在调用subscribe方法的线程上进行回调的，如果我们想切换线程就需要使用Scheduler。RxJava 已经内置了5个 Scheduler。
Schedulers.immediate()：默认的，直接在当前线程运行，相当于不指定线程。
Schedulers.newThread()：总是启用新线程，并在新线程执行操作。
Schedulers.io()：I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。
Schedulers.computation()：计算所使用的 Scheduler，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
Schedulers.trampoline()：当我们想在当前线程执行一个任务时，并不是立即时，可以用.trampoline()将它入队。这个调度器将会处理它的队列并且按序运行队列中每一个任务。

Observable.just("hello","weiwei","xiaobanma")
 .subscribeOn(Schedulers.io())
 .observerOn(AndroidSchedulers.mainThread())  //UI线程中
 .subscribe(new Action1<String>() {
      @Override
      public void call(String s){
       ｝
 }):

RxJava的基本使用获取图片

   showImg = (ImageView) findViewById(R.id.imageView);

    ///////RxJava的基本使用/////////
    Observer observer = new rx.Observer<Drawable>() {
        @Override
        public void onCompleted() { //事件队列完结 在RxJava中 不仅会把每个事件单独处理，还会把它们看成是一个队列。当没有新的事件发生时，这个时候就会触发oCompleted方法，这个方法也可以看成是事件完成标志。         Log.i(TAG,"onCompleted");
        }

        @Override
        public void onError(Throwable e) { //事件队列异常 在事件处理的过程中，难免会发生异常，在发生异常的时候，onError方法就会触发。同时事件队列就会终结
            Log.i(TAG, "onError");
        }

        @Override
        public void onNext(Drawable s) { //就是一件将要处理的普通的事件，添加到处理事件队列中
            Log.i(TAG, "onNext" + s);
            showImg.setImageDrawable(s);
        }

    });

    Observable mObServerable = Observable.create(new Observable.OnSubscribe<Drawable>() {
        @Override
        public void call(Subscriber<? super Drawable> subsciber) {

            Drawable drawable = getResources().getDrawable(R.mipmap.ic_launcher);
            subsciber.onNext(drawable);
            subsciber.onCompleted();
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new rx.Observer<Drawable>() {
        @Override
        public void onCompleted() { //事件队列完结 在RxJava中 不仅会把每个事件单独处理，还会把它们看成是一个队列。当没有新的事件发生时，这个时候就会触发oCompleted方法，这个方法也可以看成是事件完成标志。         Log.i(TAG,"onCompleted");
        }

        @Override
        public void onError(Throwable e) { //事件队列异常 在事件处理的过程中，难免会发生异常，在发生异常的时候，onError方法就会触发。同时事件队列就会终结
            Log.i(TAG, "onError");
        }

        @Override
        public void onNext(Drawable s) { //就是一件将要处理的普通的事件，添加到处理事件队列中
            Log.i(TAG, "onNext" + s);
            showImg.setImageDrawable(s);
        }

        @Override
        public void onStart() { //事件在还没发生前调用，这个方法用作一些做准备工作，比如数据的重置、清零
            Log.i(TAG, "onStart");
        }

    });
}