本文共 9201 字，大约阅读时间需要 30 分钟。

前言

刚参加工作的时候接触到了项目中的Rxjava，当时一点基础没有，学习了好长时间才渐渐学会使用，但也只是皮毛，停留在表面。后来换工作了发现Rxjava对找工作的帮助还是挺大的，因为是流行框架，都可以讲给面试官听，体现出自己追求技术的精神。但实际上，学会Rxjava对技术的提升还是很有帮助的，所以我会为大家讲解Rxjava的相关知识，以及Rxjava 1 到Rxjava 2的变化，帮助大家能够学会使用Rxjava，并能在面试中征服面试官！

本文的主要作用是：如果之前有点基础，那就是复习巩固，顺便了解一下新版本的Rxjava的变化；如果之前没有基础，那就来认识一下Rxjava,

Rxjava是什么？

RxJava 在上的说明是：

a library for composing asynchronous and event-based programs using observable sequences for the Java VM

一个在 Java VM上使用可观测的序列来组成异步的、基于事件的程序的库

这句话很难懂，但是可以概括成一个核心词——异步

Rxjava主要作用就是用来处理异步，当你的业务需要访问数据库，访问网络，或者任何耗时的操作，都可以借助Rxjava来实现。

但是有人说在Android中已经有很多异步操作的API，比如Handler，AsyncTask等，这些都能满足基本的异步操作，为什么还要使用Rxjava呢？ 首先我们开看一个例子做个比较：

假设有这样一个需求：界面上有一个自定义的视图 imageCollectorView ，它的作用是显示多张图片，并能使用 addImage(Bitmap) 方法来任意增加显示的图片。现在需要程序将一个给出的目录数组 File[] folders 中每个目录下的 png 图片都加载出来并显示在 imageCollectorView中。需要注意的是，由于读取图片的这一过程较为耗时，需要放在后台执行，而图片的显示则必须在 UI 线程执行。

常用的实现方式有多种，比如：

代码块1

//采用android自带的api实现new Thread() {    @Override    public void run() {        super.run();        for (File file : files) {            File[] files = folder.listFiles();            for (File file : files) {                if (file.getName().endsWith(".png")) {                    final Bitmap bitmap = getBitmapFromFile(file);                    getActivity().runOnUiThread(new Runnable() {                        @Override                        public void run() {                            imageCollectorView.addImage(bitmap);                        }                    });                }            }        }    }}.start();

如果使用Rxjava，则可以这样实现：

代码块2

//采用Rxjava实现Observable.from(folders)    .flatMap(new Func1
   
    >() {        @Override        public Observable
    
      call(File file) {            return Observable.from(file.listFiles());        }    })    .filter(new Func1
     
      () {        @Override        public Boolean call(File file) {            return file.getName().endsWith(".png");        }    })    .map(new Func1
      
       () {        @Override        public Bitmap call(File file) {            return getBitmapFromFile(file);        }    })    .subscribeOn(Schedulers.io())    .observeOn(AndroidSchedulers.mainThread())    .subscribe(new Action1
       
        () {        @Override        public void call(Bitmap bitmap) {            imageCollectorView.addImage(bitmap);        }    });
       
      
     
    
   

虽然代码块2的代码量比代码块1的代码量要多，但是很明显代码块2的代码看起来更整洁更优雅，而且如果读者学过Rxjava的人，会明显感觉到代码块2的可读性比代码块1的可读性要强。

由此可见Rxjava的优点即是：采用链式调用，代码简洁优雅有美感，并且可读性增强！

以上，是Rxjava的一部分优点，其实Rxjava的优点更在于它的强大。

下面我们简单了解一下Rxjava的原理：

Rxjava实现异步的方法是通过观察者模式实现的。

什么事观察者模式呢？

举个例子，用户界面可以作为一个观察者，业务数据是被观察者，用户界面观察业务数据的变化，发现数据变化后，就显示在界面上。

在Android中最常见的观察者模式是View的onClick事件模型。

如图可见，当Button持有OnClickListener对象之后，Button被点击之后会自动触发OnClickListener中的OnClick方法。

把上面的Button点击事件抽象一下就变成： 当Observable(可观察的，被观察者)的状态发生变化时，内部会通过一系列事件触发Observer(观察者)中的方法，可以做出相应的操作。 可能这样讲还是比较抽象，举个简单的生活中的例子：

以上模型中，上课铃声是被观察者，即Observable，可观察的，被观察者；学生就是观察者，即Observer（观察者）,学生听到上课铃声响了，就会去上课，这就是学生根据上课铃声所做出的反应。

也就是： 被观察者状态发生变化，观察者可以做出反应。

在Rxjava中观察者模式

RxJava 有四个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以根据情况回调来通知 Observer。

Rxjava常用的的回调方法有三种： - onNext：完成队列中的一个事件 - onComplete：完成队列中所有的事件 - onError：事件发生错误时，并且后续的事件终止。

为什么Rxjava要使用观察者模式呢？

因为观察者模式在模块之间划定了清晰的界限，降低模块耦合性，提高了代码的可维护性和重用性。

Rxjava基本使用方法

1. 创建Observer
   Observer是观察者，当被观察者状态发生变化的时候，他会收到相应的事件，使用者可以根据不同的事件进行不同的处理。

Observer
   
     observer = new Observer
    
     () {            @Override            public void onCompleted() {                Log.d("Rxjava demo", "onCompleted");            }            @Override            public void onError(Throwable e) {                Log.d("Rxjava demo", "onError");            }            @Override            public void onNext(String s) {                Log.d("Rxjava demo", "onNext");            }        };
    
   

其实，除了使用Observer以外，Rxjava还有个Subscriber。这个是实现了Observer的抽象类，里面对Observer进行了一些扩展。

Subscriber
   
     subscriber = new Subscriber
    
     () {            @Override            public void onStart() {                super.onStart();            }            @Override            public void onNext(String s) {                Log.d("Rxjava demo", "Item: " + s);            }            @Override            public void onCompleted() {                Log.d("Rxjava demo", "Completed!");            }            @Override            public void onError(Throwable e) {                Log.d("Rxjava demo", "Error!");            }        };
    
   

可以看出，Subscriber比Observer多了一个回调方法onStart()，它会在事件开始执行之前的时候调用，用于做一些准备工作，类似于AsyncTask中的onPreExecute方法。

但是subscriber中还有几个很重要的方法： - unsubscribe()：这个方法是取消订阅事件，一般有利于防止内存泄漏。在android开发中我们知道一般有订阅就应该有取消订阅。 - isUnsubscribed()：这个方法是用于判断事件是否被订阅。 - add(Subscription s)：这个方法是把一个Subscription 添加到Subscription列表中，便于统一管理，取消订阅等 2. 创建Observable

Observable observable = Observable.create(new Observable.OnSubscribe
   
    () {            @Override            public void call(Subscriber
     subscriber) {                subscriber.onNext("onNext");                subscriber.onCompleted();                subscriber.onNext("onNext");                subscriber.onError(new Throwable());            }        });
   

使用create方法创建Observable（被观察者），然后call方法会被自动调用，在call方法内部定义事件的回调的行为。

其实这段代码中，当执行了onComplete方法之后，就不会在往下执行了，也就是说onError方法不会被调用，因为事件已经完全执行完成，就会停止执行之后的事件。 如果我们反过来写：

Observable observable = Observable.create(new Observable.OnSubscribe
   
    () {            @Override            public void call(Subscriber
     subscriber) {                subscriber.onNext("onNext");                subscriber.onError(new Throwable());                subscriber.onCompleted();                subscriber.onNext("onNext");            }        });
   

onNext方法执行完成之后会执行onError，但是之后的onComplete方法以及后面的事件都不会在执行了，前面我们说过，onError执行之后表示事件执行失败，后面的事件就会停止执行。

3. Subscribe(订阅)

observable.subscribe(observer);

最后我们使用subscribe方法让observer订阅observable。但是这个方法看起来写反了，他不是“观察者”订阅“被观察者”，而是被观察者订阅了观察者，这其实是因为为了保证流式的设计，把subscribe是Observable的方法，把observer作为参数传进。

什么保证流式设计呢？ 因为Rxjava可以这样写：

Observable.create(new Observable.OnSubscribe
   
    () {            @Override            public void call(Subscriber
     subscriber) {                subscriber.onNext("onNext");                subscriber.onCompleted();                subscriber.onError(new Throwable());            }        }).subscribe(new Observer
    
     () {            @Override            public void onCompleted() {                Log.d("Rxjava demo", "onCompleted");            }            @Override            public void onError(Throwable e) {                Log.d("Rxjava demo", "onError");            }            @Override            public void onNext(String s) {                Log.d("Rxjava demo", "onNext");            }        });
    
   

这样代码就会看着优雅许多，而且层级清晰，可读性强。

通过以上方法，我们就简单了解了Rxjava的使用方法。但其实，Observable的创建方式有多种: - 例如just可以传入多个参数，最多可以传入10个参数，并且会自动调用10次onNext - from(T[])将传入的数组依次发送出去，数组内有多少个元素，就会调用多少次onNext，当所有元素（事件）发送结束之后会调用onComplete，如果在某个元素中发生错误，就会调用onError。 写法如下：

ArrayList
   
     array = new ArrayList<>();Observable.from(array).subscribe(new Subscriber
    
     () {            @Override            public void onCompleted() {            }            @Override            public void onError(Throwable e) {            }            @Override            public void onNext(String s) {            }        });
    
   

大家可以看到，不用再重写call方法，因为会自动安排事件发送，不需要手动调用onNext等方法了，而这段代码中的onNext方法会依次输出数组中的每一个元素。

Rxjava的变化

以上内容都是基于Rxjava比较旧的API介绍的，目前Rxjava 1 已经更新到了1.3

使用最新的Rxjava 1需要引入以下依赖：

compile 'io.reactivex:rxjava:1.3.0'compile 'io.reactivex:rxandroid:1.2.1'

在这个版本中的Observable的创建有所变化,方法 static Observable create(OnSubscribe f)已经过时了，因为这个方法不安全。

新的版本中已经引入了比较安全的方法： - static

Observable.create(new SyncOnSubscribe
   
    () {            @Override            protected String generateState() {                Log.d("Rxjava demo", "generateState");                return "generateState";            }            @Override            protected String next(String state, Observer
     observer) {                observer.onNext(state);                observer.onCompleted();                observer.onError(new Throwable("onError"));                return state;            }        }).subscribe(new Action1
    
     () {            @Override            public void call(String s) {                Log.d("Rxjava demo", s);            }        }, new Action1
     
      () {            @Override            public void call(Throwable throwable) {                Log.d("Rxjava demo", throwable.getMessage());            }        }, new Action0() {            @Override            public void call() {                Log.d("Rxjava demo", "onComplete");            }        });
     
    
   

但是可以看出来多了两种回调方法：

- generateState()，这个方法会在subscribe的时候调用，产生一个state值，这个值会在第一次迭代的时候传递到next(S state, Observer observer) 方法中，后续迭代下将收到由先前的调用返回下一个状态。也就是会收到next(S state, Observer observer)的返回值

• next(S state, Observer observer)中会收到上游传来的数据，并通过observer.onNext方法传递到下游。但是该方法的实现必须遵循以下规则：(1)observer.onNext(t)不能超过1次调用。(2)不能同时调用observer.onNext(t)。
  next(S state, Observer observer)会返回下一次迭代的状态值（state）给generateState()，然后generateState()再把值传递给next(S state, Observer observer)，如果你没有调用onComplete或者onError，这个循环会一直下去

好了，Rxjava简单的介绍就到这里了，下次我们会介绍Rxjava最强大的地方，也就是Rxjava操作符。