相信大家都对EventBus这个很热门的工具很熟悉了，最近更新开发新框架的缘故，接触到Retrofit 2 和 RxJava 2 ，自然而然地使用RxBus来代替EventBus。
RxJava
一辆有意思的BUS
“Bus”这个词其实很形象的说明总线这个概念。那什么是总线？
总线是系统传送信息的公共通信干线，各个部件在总线上传递消息。RxBus实现了一种事件总线，一条公共的通信通道，上面跑着各种信息，Android中的各个组件或是控件都可以向它发送各种信息，在各个组件或控件中只要订阅这条总线，每当总线收到消息的时候，这些订阅者同样就能收到这些消息。
RxBus 最基本的实现
支持注解方式订阅事件。
支持粘滞事件。
支持3种Bus:
Publish Bus( RxBus，参见 PublishSubject )
Behavior Bus(参见 BehaviorSubject)
Replay Bus(参见ReplaySubject)
这三种都是基于Subject类，它负责在非Rx-Apis之间充当桥梁。但是也存在问题。他没有能力去处理onComplete 或 onError事件.如果在下游发生onError / onComplete ，上游将受到影响。
ReRelay开源项目
JakeWharton 大神已经帮你们处理好这些问题。
Relays 是既是Observable也是Consumer的RxJava 类型，它同样能够很容易在non-Rx api和 Rx api之间搭起桥梁,而不必要担心下游的触发的终止状态（onComplete 或 onError）。
RxBus 下载使用
Maven:
<dependency>
<groupId>com.kw</groupId>
<artifactId>rxbus</artifactId>
<version>1.0.1</version>
<type>pom</type>
</dependency>
Gradle:
compile 'com.kw:rxbus:1.0.1'
这个库 并不含有 RxAndroid 和 RxJava ，请自行添加
如何使用
发送事件
RxBus.getInstance().send(new UserEvent(1,"名字"));
接受事件
//注册
disposable = RxBus.getInstance().register(UserEvent.class, AndroidSchedulers.mainThread(), new Consumer<UserEvent>() {
@Override
public void accept(UserEvent userEvent) {
btnNext.setText(userEvent.getName());
Toast.makeText(getBaseContext(), userEvent.toString(), Toast.LENGTH_SHORT).show();
Log.d("AActivity", "onNext:" + Thread.currentThread().getName());
throw new NullPointerException("空指针错误");//发生错误之后，会取消订阅
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//发生错误后仅仅会进入一次，因为发生错误之后，会取消订阅
Toast.makeText(getBaseContext(), throwable.getMessage(), Toast.LENGTH_SHORT).show();
}
});
//上下两段代码具有相同意义
disposable=RxBus.getInstance().toObservable(UserEvent.class)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<UserEvent>() {
@Override
public void accept(UserEvent userEvent) throws Exception {
btnNext.setText(userEvent.getName());
Toast.makeText(getBaseContext(),userEvent.toString(),Toast.LENGTH_SHORT).show();
Log.d("AActivity","onNext:"+Thread.currentThread().getName());
throw new NullPointerException("空指针错误");//发生错误之后，会取消订阅
}
},new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//发生错误后仅仅会进入一次，因为发生错误之后，会取消订阅
Toast.makeText(getBaseContext(),throwable.getMessage(),Toast.LENGTH_SHORT).show();
}
});
//解除
@Override
protected void onDestroy() {
super.onDestroy();
RxBus.getInstance().unregister(disposable);
}
关键代码
public final class RxBus {
private Relay<Object> bus = null;
private static RxBus instance;
//禁用构造方法
private RxBus() {
bus = PublishRelay.create().toSerialized();
}
public static RxBus getInstance() {
if (instance == null) {
synchronized (RxBus.class) {
if (instance == null) {
instance = new RxBus();
}
}
}
return instance;
}
public void send(Object event) {
bus.accept(event);
}
public <T> Observable<T> toObservable(Class<T> eventType) {
return bus.ofType(eventType);
}
...
}
使用时，需要你定义一些事件类型，例如上面的UserEvent，该类型只会传递给订阅该类型的观察者。这主要归功于ofType操作符。
这里仅仅实现了广播事件，基于ReRelay还可以实现其他方式。
如果有什么不理解的，请浏览基于Rxjava2+RxRelay 的RxBus源码。
网友评论