都说RxJava 是非常强大但是难于上手的。我接触RxJava已经有一段时间了，今天就从自己的项目中，将用到RXJava的部分单独的拿出来写一篇文章，用来帮助看了很多RxJava相关的文章但是还不知道怎么去使用的同学。
前言
阅读本文章之前，我们在回顾或者加强几个基本概念。
Observer：观察者
Observable：可观察者
Subscribe：订阅
observalbe(观察者) subscribe(订阅) observer(被观察者)
Tips
上面的逻辑看起来和我们正常的逻辑是相反的，按照常理来说不应该是被观察者订阅观察者吗？为什么反过来了，具体原因可以在 给Android开发者的 RxJava 详解 中找到答案
RxJava使用三步走
RxJava基本实现只需要三步
- 创建Observer
- 创建Observable
- 订阅
1.创建Observer
Observer即观察者，他决定事件触发的时候将会有什么样的行为。基本的Observer我们可以这么实现：
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.创建Observable
Observable 即被观察者，他决定什么时候触发怎样的事件。
我们可以使用create()方法创建一个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...)创建一个Observable
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
也可以使用from(T[])来创建一个Observable
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
3.订阅
我们创建了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对象，创建接口实例，调用接口方法，即可获取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，在业务逻辑中我想实现在查询一个单词的时候，如果本地数据库已经存在了单词记录就从本地读取记录，而不从网络获取。然而两个实体类型不同，我又想使用优雅的方法解决它，我能不能获取了YouDaoResult之后，立刻就转换成Result呢？后来我使用了RxJava的map()变换对象流方法。
Tips
在我的项目中，所有的Observable都是放在一起管理的，作为DataLayer（数据层），在业务方法中，想要获取数据首先要在数据层中获取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的优点，异步，简洁，即使逻辑复杂，已然可以保持简洁。在查询单词的业务逻辑中，主要做了下面几件事：
- 在本地数据库有单词数据时优先从本地数据库查询单词
- 本地数据库没有单词数据则从网络获取数据
- 单词在输出前进行缓存，这里又分为两步不过实现方法在数据库层。
- 异步
public void fetchTranslation(String query) {
// 分发开始刷新列表事件（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架构分发事件
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架构分发事件
Action action = new Action.Builder()
.with(TranslateActions.ACTION_TRANSLATION_NET_ERROR)
.build();
dispatcher.dispatch(action);
}
});
}
小结
RxJava并没有那么难，我们不敢将它引入到实际开发环境的最终原因只是我们对RxJava没有那么熟悉。Talk is cheap，赶紧去练习吧。
最后放上我的项目地址: Translate
欢迎围观，欢迎批评，欢迎讨论。
延伸阅读
中文学习资料：
给 Android 开发者的 RxJava 详解
过滤序列 | RxJava Essentials CN
lzyzsd/Awesome-RxJava: RxJava resources
RxJava 与 Retrofit 结合的最佳实践
