都說RxJava 是非常強大但是難于上手的。我接觸RxJava已經有一段時間了，今天就從自己的項目中，將用到RXJava的部分單獨的拿出來寫一篇文章，用來幫助看了很多RxJava相關的文章但是還不知道怎么去使用的同學。

前言

閱讀本文章之前，我們在回顧或者加強幾個基本概念。
Observer：觀察者
Observable：可觀察者
Subscribe：訂閱
observalbe(觀察者) subscribe(訂閱) observer(被觀察者)

Tips
上面的邏輯看起來和我們正常的邏輯是相反的，按照常理來說不應該是被觀察者訂閱觀察者嗎？為什么反過來了，具體原因可以在 給Android開發(fā)者的 RxJava 詳解 中找到答案

RxJava使用三步走

RxJava基本實現只需要三步

• 創(chuàng)建Observer
• 創(chuàng)建Observable
• 訂閱

1.創(chuàng)建Observer

Observer即觀察者，他決定事件觸發(fā)的時候將會有什么樣的行為?；镜腛bserver我們可以這么實現：

Observer<String> observer = new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

2.創(chuàng)建Observable

Observable 即被觀察者，他決定什么時候觸發(fā)怎樣的事件。
我們可以使用create()方法創(chuàng)建一個Observable

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});

更簡單的，我們可以使用just(T...)創(chuàng)建一個Observable

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 將會依次調用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

也可以使用from(T[])來創(chuàng)建一個Observable

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);

3.訂閱

我們創(chuàng)建了Observable和Observer之后，在用subscribe()將他們鏈接起來，代碼就可以工作啦。

observable.subscribe(observer);

在我自己的項目中RxJava使用場景舉例

RxJava與Retrofit結合

這里比較簡單，只需要稍微改變Retrofit請求接口方法的返回值類型就好了。

@GET("openapi.do?keyfrom=xxx&key=xxx&type=data&doctype=json&version=1.1")
Observable<YouDaoResult> getTranslationYouDao(@Query("q") String q);

接著使用Retrofit對象，創(chuàng)建接口實例，調用接口方法，即可獲取Observable。
我在項目中使用的Dagger2，所以看起來和只使用了RxJava與Retrofit的代碼有所不同

@Provides
@Singleton
public static ClientApi provideClientApi() {
    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(BASE_URL)
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .addConverterFactory(GsonConverterFactory.create())
            .build();
    return retrofit.create(ClientApi.class);
}

public Observable<YouDaoResult> getTranslation(String query) {
        return getApi().getTranslationYouDao(query);
    }

在我的項目中，從網絡獲取的實體類型是YouDaoResult，本地數據庫存儲的實體類型是經過簡化的Result，在業(yè)務邏輯中我想實現在查詢一個單詞的時候，如果本地數據庫已經存在了單詞記錄就從本地讀取記錄，而不從網絡獲取。然而兩個實體類型不同，我又想使用優(yōu)雅的方法解決它，我能不能獲取了YouDaoResult之后，立刻就轉換成Result呢？后來我使用了RxJava的map()變換對象流方法。

Tips
在我的項目中，所有的Observable都是放在一起管理的，作為DataLayer（數據層），在業(yè)務方法中，想要獲取數據首先要在數據層中獲取Observable，再使用RxJava的方法去處理它。

@Override
public Observable<Result> getTranslation(String query) {
    return getApi().getTranslationYouDao(query)
            .map(new Func1<YouDaoResult, Result>() {
                @Override
                public Result call(YouDaoResult youDaoResult) {
                    return youDaoResult.getResult();
                }
            });
}

在這里要放大招啦，項目中獲取單詞的方法是怎么實現的。根據代碼注釋可以很直觀的看出RxJava的優(yōu)點，異步，簡潔，即使邏輯復雜，已然可以保持簡潔。在查詢單詞的業(yè)務邏輯中，主要做了下面幾件事：

• 在本地數據庫有單詞數據時優(yōu)先從本地數據庫查詢單詞
• 本地數據庫沒有單詞數據則從網絡獲取數據
• 單詞在輸出前進行緩存，這里又分為兩步不過實現方法在數據庫層。
• 異步

public void fetchTranslation(String query) {
    // 分發(fā)開始刷新列表事件（Flux架構）
    getDispatcher().dispatch(new Action.Builder().with(TranslateActions.ACTION_TRANSLATION_LOADING).build());

    // 本地數據庫數據源
    Observable<Result> cache = getDataLayer().getTranslateService().getLocalTranslation(query);

    // 服務端數據源
    Observable<Result> network = getDataLayer().getTranslateService().getTranslation(query);

    // 沒有本地數據在使用網絡數據
    Observable<Result> source = Observable
            .concat(cache, network)
            // 依次遍歷序列中的數據源， 返回第一個符合條件的數據源
            .first(new Func1<Result, Boolean>() {
                @Override
                public Boolean call(Result result) {
                    return result != null;
                }
            });

    // 重新查詢數據則更新history列表，在save方法中有判斷，具體見TranslateDB
    source = source.doOnNext(new Action1<Result>() {
        @Override
        public void call(Result result) {
            getDataLayer().getTranslateService().saveToHistory(result);
        }
    });

    source.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Result>() {
                @Override
                public void call(Result result) {
                    // Flux架構分發(fā)事件
                    getDispatcher().dispatch(new Action.Builder()
                            .with(TranslateActions.ACTION_TRANSLATION_FINISH)
                            .bundle(TranslateActions.KEY_TRANSLATION_ANSWER, result)
                            .build());
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    // Flux架構分發(fā)事件
                    Action action = new Action.Builder()
                            .with(TranslateActions.ACTION_TRANSLATION_NET_ERROR)
                            .build();
                    dispatcher.dispatch(action);
                }
            });
}

小結

RxJava并沒有那么難，我們不敢將它引入到實際開發(fā)環(huán)境的最終原因只是我們對RxJava沒有那么熟悉。Talk is cheap，趕緊去練習吧。
最后放上我的項目地址: Translate
歡迎圍觀，歡迎批評，歡迎討論。

延伸閱讀

RxJava Github
RxJava官網

中文學習資料：
給 Android 開發(fā)者的 RxJava 詳解
過濾序列 | RxJava Essentials CN
lzyzsd/Awesome-RxJava: RxJava resources
RxJava 與 Retrofit 結合的最佳實踐