Android框架解析（4）
来自天天博客：&
EventBus的作用是发布/订阅事件总线，因为项目中用到RxJava、RxAndroid，所以完全可以使用RxJava、RxAndroid来实现EventBus。
1. 编写RxBus，用于存储所有事件Subjects。
事件是传递的最小单位，可以把任何类作为一个事件。
RxBus代码如下：
* Author: wangjie
* Email: tiantian.china.
* Date: 6/11/15.
6 public class RxBus {
private static final String TAG = RxBus.class.getSimpleName();
private static RxB
public static boolean DEBUG = false;
public static synchronized RxBus get() {
if (null == instance) {
instance = new RxBus();
private RxBus() {
private ConcurrentHashMap&Object, List&Subject&& subjectMapper = new ConcurrentHashMap&&();
@SuppressWarnings(&unchecked&)
public &T& Observable&T& register(@NonNull Object tag, @NonNull Class&T& clazz) {
List&Subject& subjectList = subjectMapper.get(tag);
if (null == subjectList) {
subjectList = new ArrayList&&();
subjectMapper.put(tag, subjectList);
Subject&T, T&
subjectList.add(subject = PublishSubject.create());
if (DEBUG) Log.d(TAG, &[register]subjectMapper: & + subjectMapper);
public void unregister(@NonNull Object tag, @NonNull Observable observable) {
List&Subject& subjects = subjectMapper.get(tag);
if (null != subjects) {
subjects.remove((Subject) observable);
if (ABTextUtil.isEmpty(subjects)) {
subjectMapper.remove(tag);
if (DEBUG) Log.d(TAG, &[unregister]subjectMapper: & + subjectMapper);
public void post(@NonNull Object content) {
post(content.getClass().getName(), content);
@SuppressWarnings(&unchecked&)
public void post(@NonNull Object tag, @NonNull Object content) {
List&Subject& subjectList = subjectMapper.get(tag);
if (!ABTextUtil.isEmpty(subjectList)) {
for (Subject subject : subjectList) {
subject.onNext(content);
if (DEBUG) Log.d(TAG, &[send]subjectMapper: & + subjectMapper);
如上述代码，RxBus只提供了register、unregister、post三个方法。
这里又加入了一个tag的概念，也可以理解为channel，注册Subject、反注册Subject和post事件的时候都需要这个tag，只有tag一致才能正常接收到事件。
比如有一个事件类HelloEvent，这个事件的作用是接收到后toast一个提示“hello”，如果两个Activity都注册了这个HelloEvent事件，但是没有tag去限制，一旦post了一个helloEvent事件后，两个Activity都会收到这个事件，导致两个Activity都会toast。如果使用tag，post这个HelloEvent的时候可以设置这个tag，只有register时也使用了这个tag才会接收到这个event。
2. 在Present（如Activity的onCreate）中注册一个Observer（以下以发送一个String类型的事件为例）
1 Observable&String& addOb = RxBus.get()
.register(&addFeedTag&, String.class);
4 addOb.observeOn(AndroidSchedulers.mainThread())
.subscribe(s -& {
// todo: Accept event and process here
如上，注册了一个String类型的事件，事件的tag是“addFeedTag”，用来增加一个Feed。使用RxAndroid在Action1中处理接收到的这个事件。
3. 在任何地方发送一个事件：
RxBus.get().post(&addFeedTag&, &hello RxBus!&);
这里发送了一个tag为“addFeedTag”的String类型的事件。
4. 反注册Observer：
RxBus.get().unregister(&addFeedTag&, addOb);
注意：这里的Tag都为“addFeedTag”。
下面使用注解的方式更简单方便地使用RxBus（嗯－。－这里才是重点）。
首先来看下使用注解后的代码：
1. 注册Observer
这一步可以省略掉。
2. 发送一个事件（这里我们换一个事件：FeedItemClickEvent，我们定义这个事件是用来处理当Feed被点击后的事件）
RxBus.get().post(new FeedItemClickEvent().setPosition(position).setFeed(feed));
3. 接收事件，然后处理
2 public void onPostAccept(Object tag, FeedItemClickEvent event) {
3 　　Logger.d(TAG, &onPostAccept event: & + event);
4 　　Feed feed = event.getFeed();
5 　　// 跳转到feed详情页面...
如上，这里只需要编写一个方法，加上Accept注解，然后在方法中进行事件处理即可。
注意：方法名可以任意
方法参数一：必须为Object类型的tag；
方法参数二，如果这个方法只接收一种事件，则写明具体的事件类型，如上；如果这个方法接收多种事件，则类型需要为Object。
4. 反注册Observer
这一步也可以省略掉。
接收多种事件：
1 @Accept(
acceptScheduler = AcceptScheduler.NEW_THREAD,
@AcceptType(tag = ActionEvent.CLOSE, clazz = String.class),
@AcceptType(tag = ActionEvent.BACK, clazz = String.class),
@AcceptType(tag = ActionEvent.EDIT, clazz = String.class),
@AcceptType(tag = ActionEvent.REFRESH, clazz = String.class)
public void onPostAccept(Object tag, Object actionEvent) {
Logger.d(TAG, &[ActionEvent]onPostAccept action event name: & + actionEvent);
// todo: Accept event and process here (in new thread)
这里@Accept注解中设置了acceptScheduler为AcceptScheduler.NEW_THREAD，指明方法运行在子线程中.
value中指明了接收的事件类型，这里表示这个方法接收4种类型的事件：CLOSE, BACK, EDIT, REFRESH.
注解解释：
@Accept注解
acceptScheduler: 指定被注解的方法运行的Scheduler。
value[]: AcceptType注解数组，用于指定接收事件的tag和class。
@AcceptType注解：
tag: 接收事件的tag
clazz: 接收事件的类型
AcceptScheduler：
详情见：rx.schedulers.Schedulers和rx.android.schedulers.AndroidSchedulers
如果设置的是AcceptScheduler.EXECUTOR或AcceptScheduler.HANDLER，则需要在Application中配置Executor和Handler：
* Author: wangjie
* Email: tiantian.china.
* Date: 6/15/15.
6 public class MyApplication extends Application {
private Executor acceptExecutor = Executors.newCachedThreadPool();
private Handler handler = new Handler(Looper.getMainLooper());
public void onCreate() {
super.onCreate();
RxBus.DEBUG = true;
DefaultAcceptConfiguration.getInstance().registerAcceptConfiguration(new DefaultAcceptConfiguration.OnDefaultAcceptConfiguration() {
public Executor applyAcceptExecutor() {
return acceptE
public Handler applyAcceptHandler() {
因为需要对Accept和AcceptType注解的解析，所以项目的BaseActivity需要使用，然后实现parserMethodAnnotations()方法，使用对注解进行解析。
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：1765045次
积分：14065
积分：14065
排名：第928名
原创：165篇
转载：215篇
评论：168条
(1)(1)(1)(2)(1)(2)(4)(4)(2)(2)(3)(2)(10)(1)(1)(1)(3)(1)(1)(7)(8)(7)(7)(1)(1)(14)(14)(2)(1)(8)(50)(6)(3)(11)(23)(25)(14)(7)(29)(24)(25)(20)(27)(9)
(window.slotbydup = window.slotbydup || []).push({
id: '4740881',
container: s,
size: '200,200',
display: 'inlay-fix'RxBus真的能替代EventBus吗？ - 简书
RxBus真的能替代EventBus吗？
本文通过原理分析和实测结果对EventBus和RxBus做详尽的对比，给大家在实际项目中选型做参考。
开始之前先提一下Otto：当初Android上的事件总线系统是两强相争，老大是EventBus，老二就是Otto。在EventBus官方的性能测试代码中，只和Otto做了对比测试。这么重要的人物为了么这次没出场呢？是因为它已经停止开发了。Otto的主页上有这么一句话：
This project is deprecated in favor of
and . These projects permit the same event-driven programming model as Otto, but they’re more capable and offer better control of threading.
If you’re looking for guidance on migrating from Otto to Rx,
is a good start.
该项目已被和取代。Rx类项目允许与Otto类似的事件驱动编程模型，而且能力更强，操作线程更方便。如果你正在寻找从Otto迁移到Rx的指导，是个很好的开始。
看，Otto临死给后起之秀RxBus投了票，所以我们只需对比EventBus和RxBus。先扯一下Bus这个词，它可以是巴士，也可以是总线。但是细一想，其本质含义都一样，都是把线路各处人或事件，经过bus，运送到他们想要去的地方。所以英语用同一个词表达这两个意思，很有道理。我们可以理解巴士就是一个运人的总线:)
开始对比之前，还是同样的方法，我们先考虑一下，一个完美的事件总线应该具备哪些功能？
容易订阅事件：事件订阅者只要声明自己就好了，当事件发生时自然会被调到。订阅和取消可以方便绑定到Activity和Fragment的生命周期上。
容易发送事件：事件发送者直接发送就好了，其他的事都不管。
方便的切换线程：有些事必须主线程干，有些事必须非主线程干，所以这个还是要说清楚。
性能：随着应用的成长，总线可能会被重度使用，性能一定要好。
我们的需求就是这么简单。
EventBus是个明星项目，github上的star超过10000，网上的分析文章很多。这里只说一下大原理，方便我们分析。详细内容可以参考。EventBus官网上说用EventBus只要三步：定义事件，创建订阅者并订阅，发送事件。我们也从这几个步骤入手。
定义事件每个事件是一个新的类，类中可以带参数，什么都不带，只是一个空的类定义，也没问题。这个设计虽然会大幅增加类的数量(我们的项目中就有100多个event类)，但是清晰易懂，相比在类中定义事件类型变量的方法，使用起来方便很多。
创建订阅者当某个事件发生时，订阅这个事件的订阅者就要被执行。所以创建订阅者时除了要完成函数本身，还要声明订阅的是哪个参数(函数参数)，并声明线程要求、是否粘性等，EventBus中这些信息通过注解声明，使用十分方便。至于如何提取和处理这些信息这样的脏活累活，EventBus都留给了自己。
@Subscribe(threadMode = ThreadMode.MAIN, sticky = true)public void onEventMainThread(StateEvent event) {
Log.d(TAG, "StateEvent is emitted");}
订阅、取消订阅
eventBus.register(this);eventBus.unregister(this);
在有订阅者的类中，随时调用register()，就把类中的所有订阅者都激活了。随时调用unregister()，就都解除订阅了。这个结合Android中Activity和Fragment的生命周期使用很方便。刚才说的脏活累活就是register()来完成的。它也是EventBus中最复杂的部分。详细分析请参见其他文章，这里只说原理。register()要做这么几件事：
通过注解把类中的所有订阅者和订阅信息提取出来，这里有很多过滤条件保证提取的正确性。
EventBus中有两个映射表，subscriptionsByEventType中存放所有event和对应的订阅者，typesBySubscriber中存放每个订阅者对应的事件。提取出来的订阅者和信息就被存进了这两个表里。
private final Map&Class&?&, CopyOnWriteArrayList&Subscription&& subscriptionsByEventTprivate final Map&Object, List&Class&?&&& typesByS
发送事件所有的的poster被放在3个队列中，mainThreadPoster、backgroundPoster、asyncPoster。这3个队列是用链表实现的，让poster在各自的线程排队，等候处理。如果是POSTING类型的事件，就直接执行了，不用排队。
post事件时，在前面的映射表中可以用事件查到所有此事件的订阅者，把每个订阅者执行一遍。执行时根据线程要求做处理。
如果是POSTING，就把函数反射出来直接执行。
如果是MAIN，就直接执行，或者放到mainThreadPoster队列中去。
如果是BACKGROUND，或者直接执行，或者放到backgroundPoster队列中去。
如果是ASYNC，放到asyncPoster队列中去。
对应的线程会执行这些poster。
EventBus加速黑科技上面就是EventBus的主要功能了。EventBus中很多代码是为了防止出错和性能优化。例如EventBus中大量使用了池来减少常用元素的创建和删除。前面提到的脏活累活有一部分放到编译期完成，例如中提到的，订阅者信息在编译期就都提取出来了，生成一个静态的SUBSCRIBER_INDEX，里面存放了类名索引的类中的订阅者信息，包含订阅的事件名、订阅者方法名、线程、是否粘性、优先级。
回过头来看我们的评价指标：
容易订阅事件：完全合格。
容易发送事件：完全合格。
方便的切换线程：完全合格。
性能：完全合格。
那么问题来了，EventBus使用方便，性能优化到位，各方面都很好，为什么会遭到RxBus的挑战，甚至可能被替代呢？
希望前面的介绍把EventBus的原理说清楚了。下面我们来看看RxBus。它不是一个库，而是一个文件，最短实现只有短短30行代码。RxBus本身不需要过多分析，它的强大完全来自于它基于的RxJava技术。响应式编程(Reactive Programming)技术这几年特别火，RxJava是它在Java上的实作。RxJava天生就是类似sub/pub的观察者模式，而且很容易处理线程切换。而EventBus做的最重要的两件事就是事件订阅和线程控制。所以才有了RxBus区区30行代码就敢挑战EventBus的老大地位。
从 Kaushik Gopal 2014年12月在中一次提出RxBus，不断有人在完善RxBus。我在实际使用中也对原始的RxBus进行了扩充，解决易用性和性能的问题。
由于RxJava已经解决了订阅和线程切换问题，RxBus尚未解决的，也是重点需要解决的是事件分发的效率问题。为此我在AndroidPlayGround中复用EventBus性能测试代码，对EventBus和RxBus做了性能测试。具体参见。测试中可以明显发现RxBus效率随订阅者增多而成比例下降。
暂时的结论
如果你的项目中已经开始使用EventBus，没有必要特意换用RxBus。
如果你的项目计划引入RxJava，并认为统一风格很重要，启用RxBus，抛弃EventBus没有问题。但是请了解RxBus的问题，并关注RxBus的进展，最好能用Rx的方式解决这个问题。
RxBus也在不断进化中，我相信随着大家的努力，我们能够解决RxBus中的问题，但是目前，RxBus在订阅者数量很大的情况下，性能下降非常明显是个事实。当RxBus有了好的解决方案，我会第一时间更新上来。
从Linux转行做Android，爱玩ROR、JS等各种技术的大爷以下内容为原创，欢迎转载，转载请注明
来自天天博客：&
EventBus的作用是发布/订阅事件总线，因为项目中用到RxJava、RxAndroid，所以完全可以使用RxJava、RxAndroid来实现EventBus。
1. 编写RxBus，用于存储所有事件Subjects。
事件是传递的最小单位，可以把任何类作为一个事件。
RxBus代码如下：
* Author: wangjie
* Email: tiantian.china.
* Date: 6/11/15.
6 public class RxBus {
private static final String TAG = RxBus.class.getSimpleName();
private static RxB
public static boolean DEBUG = false;
public static synchronized RxBus get() {
if (null == instance) {
instance = new RxBus();
private RxBus() {
private ConcurrentHashMap&Object, List&Subject&& subjectMapper = new ConcurrentHashMap&&();
@SuppressWarnings("unchecked")
public &T& Observable&T& register(@NonNull Object tag, @NonNull Class&T& clazz) {
List&Subject& subjectList = subjectMapper.get(tag);
if (null == subjectList) {
subjectList = new ArrayList&&();
subjectMapper.put(tag, subjectList);
Subject&T, T&
subjectList.add(subject = PublishSubject.create());
if (DEBUG) Log.d(TAG, "[register]subjectMapper: " + subjectMapper);
public void unregister(@NonNull Object tag, @NonNull Observable observable) {
List&Subject& subjects = subjectMapper.get(tag);
if (null != subjects) {
subjects.remove((Subject) observable);
if (ABTextUtil.isEmpty(subjects)) {
subjectMapper.remove(tag);
if (DEBUG) Log.d(TAG, "[unregister]subjectMapper: " + subjectMapper);
public void post(@NonNull Object content) {
post(content.getClass().getName(), content);
@SuppressWarnings("unchecked")
public void post(@NonNull Object tag, @NonNull Object content) {
List&Subject& subjectList = subjectMapper.get(tag);
if (!ABTextUtil.isEmpty(subjectList)) {
for (Subject subject : subjectList) {
subject.onNext(content);
if (DEBUG) Log.d(TAG, "[send]subjectMapper: " + subjectMapper);
如上述代码，RxBus只提供了register、unregister、post三个方法。
这里又加入了一个tag的概念，也可以理解为channel，注册Subject、反注册Subject和post事件的时候都需要这个tag，只有tag一致才能正常接收到事件。
比如有一个事件类HelloEvent，这个事件的作用是接收到后toast一个提示&hello&，如果两个Activity都注册了这个HelloEvent事件，但是没有tag去限制，一旦post了一个helloEvent事件后，两个Activity都会收到这个事件，导致两个Activity都会toast。如果使用tag，post这个HelloEvent的时候可以设置这个tag，只有register时也使用了这个tag才会接收到这个event。
2. 在Present（如Activity的onCreate）中注册一个Observer（以下以发送一个String类型的事件为例）
1 Observable&String& addOb = RxBus.get()
.register("addFeedTag", String.class);
4 addOb.observeOn(AndroidSchedulers.mainThread())
.subscribe(s -& {
// todo: Accept event and process here
如上，注册了一个String类型的事件，事件的tag是&addFeedTag&，用来增加一个Feed。使用RxAndroid在Action1中处理接收到的这个事件。
3. 在任何地方发送一个事件：
RxBus.get().post("addFeedTag", "hello RxBus!");
这里发送了一个tag为&addFeedTag&的String类型的事件。
4. 反注册Observer：
RxBus.get().unregister("addFeedTag", addOb);
注意：这里的Tag都为&addFeedTag&。
下面使用注解的方式更简单方便地使用RxBus（嗯－。－这里才是重点）。
首先来看下使用注解后的代码：
1. 注册Observer
这一步可以省略掉。
2. 发送一个事件（这里我们换一个事件：FeedItemClickEvent，我们定义这个事件是用来处理当Feed被点击后的事件）
RxBus.get().post(new FeedItemClickEvent().setPosition(position).setFeed(feed));
3. 接收事件，然后处理
2 public void onPostAccept(Object tag, FeedItemClickEvent event) {
3 　　Logger.d(TAG, "onPostAccept event: " + event);
4 　　Feed feed = event.getFeed();
5 　　// 跳转到feed详情页面...
如上，这里只需要编写一个方法，加上Accept注解，然后在方法中进行事件处理即可。
注意：方法名可以任意
方法参数一：必须为Object类型的tag；
方法参数二，如果这个方法只接收一种事件，则写明具体的事件类型，如上；如果这个方法接收多种事件，则类型需要为Object。
4. 反注册Observer
这一步也可以省略掉。
接收多种事件：
1 @Accept(
acceptScheduler = AcceptScheduler.NEW_THREAD,
@AcceptType(tag = ActionEvent.CLOSE, clazz = String.class),
@AcceptType(tag = ActionEvent.BACK, clazz = String.class),
@AcceptType(tag = ActionEvent.EDIT, clazz = String.class),
@AcceptType(tag = ActionEvent.REFRESH, clazz = String.class)
public void onPostAccept(Object tag, Object actionEvent) {
Logger.d(TAG, "[ActionEvent]onPostAccept action event name: " + actionEvent);
// todo: Accept event and process here (in new thread)
这里@Accept注解中设置了acceptScheduler为AcceptScheduler.NEW_THREAD，指明方法运行在子线程中.
value中指明了接收的事件类型，这里表示这个方法接收4种类型的事件：CLOSE, BACK, EDIT, REFRESH.
注解解释：
@Accept注解
acceptScheduler: 指定被注解的方法运行的Scheduler。
value[]: AcceptType注解数组，用于指定接收事件的tag和class。
@AcceptType注解：
tag: 接收事件的tagclazz: 接收事件的类型
AcceptScheduler：
详情见：rx.schedulers.Schedulers和rx.android.schedulers.AndroidSchedulers
如果设置的是AcceptScheduler.EXECUTOR或AcceptScheduler.HANDLER，则需要在Application中配置Executor和Handler：
* Author: wangjie
* Email: tiantian.china.
* Date: 6/15/15.
6 public class MyApplication extends Application {
private Executor acceptExecutor = Executors.newCachedThreadPool();
private Handler handler = new Handler(Looper.getMainLooper());
public void onCreate() {
super.onCreate();
RxBus.DEBUG = true;
DefaultAcceptConfiguration.getInstance().registerAcceptConfiguration(new DefaultAcceptConfiguration.OnDefaultAcceptConfiguration() {
public Executor applyAcceptExecutor() {
return acceptE
public Handler applyAcceptHandler() {
因为需要对Accept和AcceptType注解的解析，所以项目的BaseActivity需要使用，然后实现parserMethodAnnotations()方法，使用对注解进行解析。
阅读(...) 评论()使用RxJava代替EventBus类库 - 安卓 - 伯乐在线
& 使用RxJava代替EventBus类库
如今的Android社区，人人都在讨论RxJava以及为什么我们应该在项目中使用RxJava。当我们开始在Android项目中使用RxJava的时候，就已经意识到了，我们的代码库可以不再需要Otto了（或其他事件总线类库）。
使用MVP构建代码库
当我们在开发一款叫做Radyoland的无线流媒体应用的时候，我们决定使用MVP模式来设计我们的代码库和项目架构等。于是我们把它分为几个层（domain, model, app等等）。
在model层中，存在一些类和接口用来处理RESTful。而domain层中，我们试图实现应用的业务逻辑，因此创建了一些usecase类。
为什么需要事件总线类库？
如果你的Android程序中有超过一个逻辑层，就意味着你有可能在层与层之间进行数据的传递。在我们的例子中，我们认为，为DataBus和UIBus创建一个BusUtil就能够轻松实现层与层之间的数据传递（model, domain, presentation）。
你可以“订阅”或者“取消订阅”从bus中所发送出的具体事件。这个方法的工作原理看起来就是这样。
在UsecaseController，PresenterImp 类之间，我们把REST实现类中得到的结果作为事件发送，然后订阅此事件。
@Override public void getRadioList() {
Call radioWrapperCall = restInterface.getRadioList();
radioWrapperCall.enqueue(new Callback() {
@Override public void onResponse(Response response, Retrofit retrofit) {
dataBus.post(response.body());
@Override public void onFailure(Throwable t) {
Timber.e(t.getMessage());
123456789101112
@Override public void getRadioList() {&&&&Call radioWrapperCall = restInterface.getRadioList();&&&&radioWrapperCall.enqueue(new Callback() {&&&&&&@Override public void onResponse(Response response, Retrofit retrofit) {&&&&&&&&dataBus.post(response.body());&&&&&&}&&&&&&&@Override public void onFailure(Throwable t) {&&&&&&&&Timber.e(t.getMessage());&&&&&&}&&&&});&&}
当我们通过回调函数进行异步请求的时候，使用bus发送请求成功后的结果，然后订阅这个结果事件。
@Subscribe @Override public void onRadioListLoaded(RadioWrapper radioWrapper) {
new SaveDatabase(radioWrapper, databaseSource).executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR);
uiBus.post(radioWrapper);
@Subscribe @Override public void onRadioListLoaded(RadioWrapper radioWrapper) {&&&&new SaveDatabase(radioWrapper, databaseSource).executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR);&&&&uiBus.post(radioWrapper);&&}
然后再次发送这个事件至更新UI。Activity 或者Fragment 应该在onResume()方法中订阅需要的事件，相反地，在onPause()中对不需要的事件解除订阅。
@Subscribe @Override public void onRadioListLoaded(RadioWrapper radioWrapper) {
radioListView.onListLoaded(radioWrapper);
radioListView.dismissLoading();
@Subscribe @Override public void onRadioListLoaded(RadioWrapper radioWrapper) {&&&&radioListView.onListLoaded(radioWrapper);&&&&radioListView.dismissLoading();&&}
虽然，这种方案很有效。但是，作为优秀的Android工程师，应该尝试去发现更好的实现思路。因此，我们找到了一种摆脱所有回调函数和订阅方法的思路。那就是在Android应用的代码库中引入并使用RxJava和RxAndroid。
如何使用RxJava
首先，我们需要更改所有REST接口的返回类型。
不使用RxJava：
Call getRadioList();
@GET("")Call getRadioList();
使用RxJava：
Observable getRadioList();
@GET("")Observable getRadioList();
在REST实现类中，我们持有很多API调用方法。在这里，我只举出其中一例。开始使用RxJava之后，我们需要修改所有方法的实现方式。返回类型将变为Observable，当做完必要的修改之后，方法看起来如下：
@RxLogObservable @Override public Observable getRadioList() {
return restInterface.getRadioList();
@RxLogObservable @Override public Observable getRadioList() {&&&&return restInterface.getRadioList();&&}
我们已经完成了REST实现类的修改。接下来，我们要做的是修改usecase实现类和其中的方法。
@Override public Observable getRadioList() {
* Get radiolist observable from Cache
Observable radioListDB = databaseSource.getRadioList().filter(new Func1() {
@Override public Boolean call(RadioWrapper radioWrapper) {
return radioWrapper.radioList.size() & 0;
}).putation());
* Load radiolist from api layer and save it to DB.
Observable radioListApi = apiSource.getRadioList().doOnNext(new Action1() {
@Override public void call(final RadioWrapper radioWrapper) {
Observable.create(new Observable.OnSubscribe() {
@Override public void call(Subscriber super Object& subscriber) {
databaseSource.save(radioWrapper);
subscriber.onCompleted();
}).putation()).subscribe();
}).subscribeOn(Schedulers.io());
* concat db and api observables
return Observable.concat(radioListDB, radioListApi).observeOn(AndroidSchedulers.mainThread());
123456789101112131415161718192021222324252627282930
@Override public Observable getRadioList() {&&&&&/**&&&& * Get radiolist observable from Cache&&&& */&&&&Observable radioListDB = databaseSource.getRadioList().filter(new Func1() {&&&&&&@Override public Boolean call(RadioWrapper radioWrapper) {&&&&&&&&return radioWrapper.radioList.size() & 0;&&&&&&}&&&&}).subscribeOn(Schedulers.computation());&&&&&/**&&&& * Load radiolist from api layer and save it to DB.&&&& */&&&&Observable radioListApi = apiSource.getRadioList().doOnNext(new Action1() {&&&&&&@Override public void call(final RadioWrapper radioWrapper) {&&&&&&&&Observable.create(new Observable.OnSubscribe() {&&&&&&&&&&@Override public void call(Subscriber super Object& subscriber) {&&&&&&&&&&&&databaseSource.save(radioWrapper);&&&&&&&&&&&&subscriber.onCompleted();&&&&&&&&&&}&&&&&&&&}).subscribeOn(Schedulers.computation()).subscribe();&&&&&&}&&&&}).subscribeOn(Schedulers.io());&&&&&/**&&&& * concat db and api observables&&&& */&&&&return Observable.concat(radioListDB, radioListApi).observeOn(AndroidSchedulers.mainThread());&&}
现在，getRadioList()方法将返回Observable数据流传至UI。就如你所见到的一样，我们不再使用Event bus来发送事件了。可以通过对数据流的过滤，合并，缓存或者其他操作，来达到我们的目的了。
我们学到了什么
尽管RxJava用起来不是那么容易，但是当用RxJava替换掉Otto后，我们从代码库中成功移除了很多回调代码块。依我看来，RxJava最棒的地方就是能够对任何REST API进行异步请求。
可能感兴趣的话题
关于安卓频道
安卓频道分享Android开发文章，精选工具和安卓相关的行业动态。
新浪微博：
推荐微信号
（加好友请注明来意）
– 好的话题、有启发的回复、值得信赖的圈子
– 分享和发现有价值的内容与观点
– 为IT单身男女服务的征婚传播平台
– 优秀的工具资源导航
– 翻译传播优秀的外文文章
– 国内外的精选文章
– UI,网页，交互和用户体验
– 专注iOS技术分享
– 专注Android技术分享
– JavaScript, HTML5, CSS
– 专注Java技术分享
– 专注Python技术分享
& 2017 伯乐在线