Android响应式编程（一）RxJava前篇[入门基础]

本文首发于微信公众号「后厂村码农」

1.RxJava概述

ReactiveX与RxJava

在讲到RxJava之前我们首先要了解什么是ReactiveX，因为RxJava是ReactiveX的一种java实现。
ReactiveX是Reactive Extensions的缩写，一般简写为Rx，微软给的定义是，Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流， 用Schedulers参数化异步数据流的并发处理，Rx可以这样定义：Rx = Observables + LINQ + Schedulers。

为何要用RxJava

想到异步的操作我们会想到android的AsyncTask 和Handler，但是随着请求的数量越来越多，代码逻辑将会变得越来越复杂而RxJava却仍旧能保持清晰的逻辑。RxJava的原理就是创建一个Observable对象来干活，然后使用各种操作符建立起来的链式操作，就如同流水线一样把你想要处理的数据一步一步地加工成你想要的成品然后发射给Subscriber。

RxJava与观察者模式

RxJava的异步操作是通过扩展的观察者模式来实现的，不了解观察者模式的可以先看下 设计模式（五）观察者模式这篇文章Rxjava有四个基本的要素：Observable (被观察者)、 Observer (观察者)、 subscribe (订阅)、event(事件)。Observable (被观察者) 和 Observer (观察者)通过 subscribe() 方法实现订阅关系，Observable就可以在需要的时候来通知Observer。

2.RxJava基本用法

在使用RxJava前请现在Android Studio 配置gradle：

dependencies {
    ...
    compile 'io.reactivex:rxjava:1.1.6'
    compile 'io.reactivex:rxandroid:1.2.1'
}

其中RxAndroid是RxJava的一部分，在普通的RxJava基础上添加了几个有用的类，比如特殊的调度器，后文会提到。

RxJava的基本用法分为三个步骤，他们分别是：

创建Observer（观察者）

决定事件触发的时候将有怎样的行为

   Subscriber subscriber=new Subscriber<String>() {
    @Override
    public void onCompleted() {
        Log.i("wangshu","onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        Log.i("wangshu","onError");
    }

    @Override
    public void onNext(String s) {
        Log.i("wangshu","onNext"+s);
    }

    @Override
    public void onStart() {
        Log.i("wangshu","onStart");
    }
};

其中onCompleted、onError和onNext是必须要实现的方法，他们的含义分别是：

• onCompleted：事件队列完结，RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。当不会再有新的 onNext发出时，需要触发 onCompleted() 方法作为完成标志。
• onError：事件队列异常，在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
• onNext：普通的事件，将要处理的事件添加到事件队列中。
• onStart：它会在事件还未发送之前被调用，可以用于做一些准备工作。例如数据的清零或重置，这是一个可选方法，默认情况下它的实现为空。

当然如果要实现简单的功能也可以用到Observer来创建观察者，Observer是一个接口，而上面用到Subscriber是在Observer基础上进行了扩展，在后文的Subscribe订阅过程中Observer也会先被转换为Subscriber来使用。

Observer<String> observer = new Observer<String>() {
    @Override
    public void onCompleted() {
        Log.i("wangshu", "onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        Log.i("wangshu", "onError");
    }

    @Override
    public void onNext(String s) {
        Log.i("wangshu", "onNext" + s);
    }
};

创建 Observable(被观察者)

它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则：

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            subscriber.onNext("杨影枫");
            subscriber.onNext("月眉儿");
            subscriber.onCompleted();
        }
    });

通过调用subscriber的方法，不断的将事件添加到任务队列中，也可用just来实现：

Observable observable = Observable.just("杨影枫", "月眉儿");

上述的代码会依次调用onNext(“杨影枫”)、onNext(“月眉儿”)、onCompleted()。

Subscribe (订阅)

订阅比较简单：

observable.subscribe(subscriber);

或者也可以调用

observable.subscribe(observer);

运行代码查看log:

com.example.liuwangshu.moonrxjava I/wangshu: onStart
com.example.liuwangshu.moonrxjava I/wangshu: onNext杨影枫
com.example.liuwangshu.moonrxjava I/wangshu: onNext月眉儿
com.example.liuwangshu.moonrxjava I/wangshu: onCompleted

3.不完整定义回调

上文介绍了回调的接收主要是依赖subscribe(Observer) 和 subscribe(Subscriber)，除此之外RxJava还提供了另一种回调方式，也就是不完整回调。再讲到不完整回调之前我们首先要了解Action，查看RxJava源码我们发现提供了一堆Action：

我们打开Action0来看看：

public interface Action0 extends Action {
    void call();
}

再打开Action1：

public interface Action1<T> extends Action {
    void call(T t);
}

最后看看Action9：

public interface Action9<T1, T2, T3, T4, T5, T6, T7, T8, T9> extends Action {
    void call(T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6, T7 t7, T8 t8, T9 t9);
}

很明显Action后的数字代表回调的参数类型数量，上文订阅也就可以改写为下面的代码：

Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        Log.i("wangshu", "onNext" + s);
    }
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    @Override
    public void call(Throwable throwable) {

    }
};
Action0 onCompletedAction = new Action0() {
    @Override
    public void call() {
        Log.d("wangshu", "onCompleted");
    }
};
observable.subscribe(onNextAction,onErrorAction,onCompletedAction);

我们定义了onNextAction来处理onNext的回调，同理我们还定义了onErrorAction和onCompletedAction，最后我们把他传给subscribe方法。很显然这样写的灵活度很大一些，同时我们也可以只传一个或者两个Action：

observable.subscribe(onNextAction);
observable.subscribe(onNextAction,onErrorAction);

第一行只定义了onNextAction来处理onNext的回调，而第二行则定义了onNextAction处理onNext的回调，onErrorAction来处理onError的回调。

**4.Scheduler **

内置的Scheduler

方才我们所做的都是运行在主线程的，如果我们不指定线程，默认是在调用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()将它入队。这个调度器将会处理它的队列并且按序运行队列中每一个任务。

另外RxAndroid也提供了一个常用的Scheduler：

• AndroidSchedulers.mainThread()：RxAndroid库提供的Scheduler，它指定的操作在主线程中运行。

控制线程

subscribeOn() 和 observeOn() 两个方法来对线程进行控制。
subscribeOn()方法指定 subscribe() 这个方法所在的线程，即事件产生的线程。observeOn()方法指定 Subscriber 回调所运行在的线程，即事件消费的线程。

Action1<String> onNextAction = new Action1<String>() {
            @Override
            public void call(String s) {
                Log.i("wangshu", "onNext" + s);

            }
        };
Observable observable = Observable.just("杨影枫", "月眉儿"); 
            observable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()).subscribe(onNextAction);

我们仍旧是用log打印出onNext事件所传递过来的字符串，只不过这一次事件的产生的线程是在io线程上，事件回调的线程则是在主线程。

###5.**RxJava基础应用 **
好了，讲的不是很多，我们来举一个例子来消化上面的知识。RxJava+Retrofit访问网络是比较搭的，但是此前我的网络系列并没有介绍Retrofit，所以我们先准备用RxJava+OKHttp来访问网络，至于RxJava+Retrofit访问网络会在此系列的以后的章节做介绍。OKHttp的用法请详见Android网络编程（六）OkHttp3用法全解析这篇文章。
此前我们用OkHttp3访问网络是这样做的：

  private void postAsynHttp(int size) {
    mOkHttpClient=new OkHttpClient();
    RequestBody formBody = new FormBody.Builder()
            .add("size", size+"")
            .build();
    Request request = new Request.Builder()
            .url("http://api.1-blog.com/biz/bizserver/article/list.do")
            .post(formBody)
            .build();
    Call call = mOkHttpClient.newCall(request);
    call.enqueue(new Callback() {
        @Override
        public void onFailure(Call call, IOException e) {

        }
        @Override
        public void onResponse(Call call, Response response) throws IOException {
            String str = response.body().string();
            Log.i("wangshu", str);
            runOnUiThread(new Runnable() {
                @Override
                public void run() {
                    Toast.makeText(getApplicationContext(), "请求成功", Toast.LENGTH_SHORT).show();
                }
            });
        }

    });
}

接下来我们进行改造，首先我们创建Observable（被观察者）：

  private Observable<String> getObservable(final int size){
    Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(final Subscriber<? super String> subscriber) {
            mOkHttpClient=new OkHttpClient();
            RequestBody formBody = new FormBody.Builder()
                    .add("size",size+"")
                    .build();
            Request request = new Request.Builder()
                    .url("http://api.1-blog.com/biz/bizserver/article/list.do")
                    .post(formBody)
                    .build();
            Call call = mOkHttpClient.newCall(request);
            call.enqueue(new Callback() {
                @Override
                public void onFailure(Call call, IOException e) {
                    subscriber.onError(new Exception("error"));
                }

                @Override
                public void onResponse(Call call, Response response) throws IOException {
                    String str = response.body().string();
                    subscriber.onNext(str);
                    subscriber.onCompleted();
                }
            });
        }
    });
 return observable;
}

我们将根据Okhttp的回调（不在主线程）来定义事件的规则，调用subscriber.onNext来将请求返回的数据添加到事件队列中。接下来我们来实现观察者：

private void postAsynHttp(int size){   
getObservable(size).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<String>() {
           @Override
           public void onCompleted() {
               Log.i("wangshu", "onCompleted");
           }

           @Override
           public void onError(Throwable e) {
             Log.i("wangshu", e.getMessage());
           }

           @Override
           public void onNext(String s) {
               Log.i("wangshu", s);
               Toast.makeText(getApplicationContext(), "请求成功", Toast.LENGTH_SHORT).show();
           }
       });
   }

我们将访问网络回调设置为主线程，所以Toast是能正常显示的。好了这一篇就讲到这里，关于RxJava的文章后期还会写，敬请期待。

参考资料
[译] ReactiveX 的理念和特点
RxJava快速入门
给 Android 开发者的 RxJava 详解
谜之RxJava （三）update 2 —— subscribeOn 和 observeOn 的区别
RxJava开发精要3-向响应式世界问好
详细解析RxAndroid的使用方式
RxJava学习笔记2：基于RxJava+okHttp的Rest Cas登录实现
RxJava（01-介绍与初体验）