RXJava
阅读全文: http://gitbook.cn/gitchat/activity/5ebe5867ef4eff0c0bf6f277
您还可以下载 CSDN 旗下精品原创内容社区 GitChat App ，阅读更多 GitChat 专享技术内容哦。

RxJava

RxJava

Rx是Reactive Extensions的簡寫，翻譯為響應的擴展。也就是通過由一方發出信息，另一方響應信息並作出處理的核心框架代碼。該框架由微軟的架構師Erik Meijer領導的團隊開發，並在2012年11月開源。
Rx庫支持.NET、JavaScript和C++等，現在已經支持幾乎全部的流行程式語言了。
Rx的大部分語言庫由ReactiveX這個組織負責維護，比較流行的有RxJava/RxJS/Rx.NET，社區網站是reactivex.io。
RxJava作為一個流行的框架，其源碼依託在GitHub,除了支持RxJava，針對安卓系統也除了一個支持框架RxAndroid

使用時機

異步處理
為何使用

與AysncTask與Handler相較下程式碼簡潔，可讀性高，可控制的狀態多
使用方式

使用Grandle方式加入依賴Rxjava依賴:
//RxJava
compile 'io.reactivex.rxjava2:rxjava:2.1.5'
//RxAndroid
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

RxJava基本用法
1.創建觀察者Observer
Observer<String> observer = new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull String s) {
        Logger.d(s);
    }
    @Override
    public void onError(@NonNull Throwable e) {
        Logger.d(e.getMessage());
    }
    @Override
    public void onComplete() {
        Logger.d("***onComplete***");
    }
};

2.創建被觀察者Observable
Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter e) throws Exception {
        //加入兩個事件
        e.onNext("dayongxin");
        e.onNext("DoubleDa");
        e.onComplete();
    }
});

3.執行與創建訂閱
observable.subscribe(observer);

4.整合使用範例
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
        e.onNext("dayongxin");
        e.onNext("DoubleDa");
        e.onComplete();
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull String s) {
        Logger.d(s);
    }
    @Override
    public void onError(@NonNull Throwable e) {
        Logger.d(e.getMessage());
    }
    @Override
    public void onComplete() {
        Logger.d("***onComplete***");
    }
});

一些常用操作符----
5.定時Task

建立一個按照固定時間間格發射整數序列的Observable，相當於定時器。
Observable.interval(5, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull Long aLong) {
        Logger.d(aLong);
    }
    @Override
    public void onError(@NonNull Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

6.zip

zip操作碼並且兩個或多個Observable發射出的數據項，根據指定的函數變換它們，並發射一個新值。
Observable<String> observable1 = Observable.just("a", "b", "c");
Observable<Integer> observable2 = Observable.just(1, 2, 3);
Observable.zip(observable1, observable2, new BiFunction<String, Integer, String>() {
    @Override
    public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
        return s + integer;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull String s) {
       //Do something..
    }
    @Override
    public void onError(@NonNull Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

7.Scheduler
Scheduler：線程控制器，可以指定每一段代碼在什麼樣的線程中行為。

在RxJava中，存在一個未指定的線程情況，默認會使用當前線程，若是操作中有耗时的操作，會導致線程阻塞。
如何使用Scheduler控制線程:

Schedulers.immediate(): 直接在當前線程運行，相當於不指定線程。這是默認的 Scheduler。
Schedulers.newThread(): 總是啟用新線程，並在新線程執行操作。
Schedulers.io(): I/O 操作（讀寫文件、讀寫數據庫、網絡信息交互等）所使用的 Scheduler。行為模式和

newThread() 差不多，區別在於 io() 的內部實現是是用一個無數量上限的線程池，可以重用空閒的線程，因此多數情況下 io() 比 newThread() 更有效率。不要把計算工作放在 io() 中，可以避免創建不必要的線程。
Schedulers.computation(): 計算所使用的 Scheduler。這個計算指的是 CPU 密集型計算，即不會被

I/O 等操作限制性能的操作，例如圖形的計算。這個 Scheduler 使用的固定的線程池，大小為 CPU 核數。不要把 I/O

操作放在 computation() 中，否則 I/O 操作的等待時間會浪費 CPU。
Android 還有一個專用的 AndroidSchedulers.mainThread()，它指定的操作將在 Android

主線程運行。

Sample:
  private void rxHelloWord() {
        Observable.just("Hello", "RxWord!")
                .subscribeOn(Schedulers.newThread())//指定 subscribe() 發生在新的線程
                .observeOn(AndroidSchedulers.mainThread())// 指定 Subscriber 的回調發生在主線程
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, s);
                    }
                });

前言
本文基于RxJava、Retrofit的使用，若是对RxJava或Retrofit还不了解的简友可以先了解RxJava、Retrofit的用法再来看这篇文章。

在这片文章之前分别单独介绍过Rxjava以及Retrofit的使用：

Android Retrofit 2.0 的使用

Android RxJava的使用（一）基本用法

（以及后面的几篇，就不一一列出了）
使用
在了解了RxJava和Retrofit分别的用法后，RxJava、Retrofit的搭配使用也就不再话下了。

先看看使用Retrofit完成一次网络请求是怎样的

单独使用Retrofit

1、先写一个service
interface MyService {
    @GET("user/login" )
    Call<UserInfo> login(
            @Query("username") String username,
            @Query("password") String password
    );
}
2、获取Call执行网络请求
        Retrofit retrofit = new Retrofit.Builder()
                .addConverterFactory(GsonConverterFactory.create())
                .baseUrl(BASE_URL)
                .build();
        MyService service = retrofit.create(MyService.class);

        Call<UserInfo> call = service.login("1111", "ssss");
        call.enqueue(new Callback<UserInfo>() {
            @Override
            public void onResponse(Call<UserInfo> call, Response<UserInfo> response) {
                //请求成功操作
            }
            @Override
            public void onFailure(Call<UserInfo> call, Throwable t) {
                //请求失败操作
            }
        });
以上是Retrofit单独使用时的做法。那Retrofit与RxJava结合是怎样使用的？下面就来说说这篇文章的重点。
RxJava + Retrofit完成网络请求
1、添加依赖。前四个分别是RxJava、RxAndroid、Retrofit以及Gson的库，最后那个才是新加入的，RxJava + Retrofit的使用需要用到最后那个包。
    compile 'io.reactivex:rxjava:x.y.z'
    compile 'io.reactivex:rxandroid:1.0.1'
    compile 'com.squareup.retrofit2:retrofit:2.0.2'
    compile 'com.squareup.retrofit2:converter-gson:2.0.2'
    compile 'com.squareup.retrofit2:adapter-rxjava:2.0.2'
注意：最后三个包的版本号必须一样，这里用的是2.0.2。
2、写一个登录的service
interface MyService {
    @GET("user/login" )
    Observable<UserInfo> login(
            @Query("username") String username,
            @Query("password") String password
    );
}
相比之前的service，这里getNews方法的返回值是Observable类型。Observable...是不是觉得很熟悉，这货不就是之前在RxJava使用到的被监听者？
3、使用Observable完成一个网络请求，登录成功后保存数据到本地。
        Retrofit retrofit = new Retrofit.Builder()
              .addConverterFactory(GsonConverterFactory.create())
              .addCallAdapterFactory(RxJavaCallAdapterFactory.create())//新的配置
              .baseUrl(BASE_URL)
              .build();
        MyService service = retrofit.create(MyService.class);

        service.login(phone, password)               //获取Observable对象
                .subscribeOn(Schedulers.newThread())//请求在新的线程中执行
                .observeOn(Schedulers.io())         //请求完成后在io线程中执行
                .doOnNext(new Action1<UserInfo>() {
                    @Override
                    public void call(UserInfo userInfo) {
                        saveUserInfo(userInfo);//保存用户信息到本地
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())//最后在主线程中执行
                .subscribe(new Subscriber<UserInfo>() {
                    @Override
                    public void onCompleted() {
                        
                    }

                    @Override
                    public void onError(Throwable e) {
                        //请求失败
                    }

                    @Override
                    public void onNext(UserInfo userInfo) {
                        //请求成功
                    }
                });
RxJava + Retrofit 形式的时候，Retrofit 把请求封装进 Observable ，在请求结束后调用 onNext() 或在请求失败后调用 onError()。
可以看到，调用了service的login方法后得到Observable对象，在新的线程中执行网络请求，请求成功后切换到io线程执行保存用户信息的动作，最后再切换到主线程执行请求失败onError()、请求成功onNext()。整体的逻辑十分清晰都在一条链中，就算还有别的要求还可以往里面添加，丝毫不影响代码的简洁。（终于举了一个有实际意义的例子）
注意：retrofit的初始化加了一行代码
addCallAdapterFactory(RxJavaCallAdapterFactory.create())
RxJava + Retrofit 进阶
在上面举到登录后保存用户信息的例子，其实在做项目的时候，往往在登录后得到的并不是用户信息。一般登录后会得到token，然后根据token去获取用户的信息。他们的步骤是这样的：
1、登录
2、获取用户信息（前提：登录成功）
可以看得出来，这是一个嵌套的结构...嵌套啊！！！天呐，最怕嵌套的结构了。
使用RxJava + Retrofit来完成这样的请求（借用抛物线的例子，稍微做了点改动）
 //登录，获取token
@GET("/login")
public Observable<String> login(   
    @Query("username") String username,
    @Query("password") String password);
 //根据token获取用户信息
@GET("/user")
public Observable<User> getUser(
    @Query("token") String token);
//..................................
service.login("11111", "22222")
    .flatMap(new Func1<String, Observable<User>>() {  //得到token后获取用户信息
        @Override
        public Observable<User> onNext(String token) {
            return service.getUser(token);
        })
    .subscribeOn(Schedulers.newThread())//请求在新的线程中执行请求
    .observeOn(Schedulers.io())         //请求完成后在io线程中执行
    .doOnNext(new Action1<User>() {      //保存用户信息到本地
         @Override
         public void call(User userInfo) {
             saveUserInfo(userInfo);
         }
     })
    .observeOn(AndroidSchedulers.mainThread())//在主线程中执行
    .subscribe(new Observer<User>() {
        @Override
        public void onNext(User user) {
            //完成一次完整的登录请求
            userView.setUser(user);
        }

        @Override
        public void onCompleted() {
  
        }

        @Override
        public void onError(Throwable error) {
            //请求失败
        }
    });