制作简单的RxBus实现事件总线
自从接触了Rxjava之后就一直在使用Rxjava的方式来修改以前写过的一些代码，前段时间实现了Retrofit2+Rxjava的方式来请求数据感觉还不错，那么既然使用了这种方式来获取数据，自然也可以使用rx的方式来对数据进行处理。
之前我们可能使用callback、handler、Otto、EventBus...的方式处理数据，现在我们就来看下怎样自己制作简单的rxbus实现对数据的处理。
备注：rxjava已经升级为2.0下面是升级为2.0的RxBus传送门
http://www.jianshu.com/p/b22e6b10c6cf
1 使用方式
1.1 注册与取消注册
在合适的地方进行注册和取消注册
RxBus.getInstance().register(this);
RxBus.getInstance().unRegister(this);
@Override
protected void onCreate(@Nullable Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
RxBus.getInstance().register(this);
}
@Override
protected void onDestroy() {
super.onDestroy();
RxBus.getInstance().unRegister(this);
}
1.2 发布事件与订阅事件
//发布
RxBus.getInstance().post(0, s.getBookses());
//订阅
@Subscribe(tag = 0, thread = EventThread.MAIN_THREAD)
private void dataBinding(ArrayList<TravelNoteBook.Books> bookses) {
...//处理逻辑
}
当数据被发布之后，如果有订阅者订阅，就会获取到发布过的数据。这也是一个最简单的rxbus，那么我们就来看一下rxbus背后的处理，后面的内容最好对rxjava和lambda方面的知识有一些了解。
2 实现最简单的rxbus
对于上面用注解方式来订阅事件的方式,我们先用最简单的方式实现，然后把注解方式推导出来
2.1 发布与订阅
首先我们要有一个Subject 来处理发布和订阅事件
private final Subject bus;
/**
* PublishSubject 创建一个可以在订阅之后把数据传输给订阅者Subject
* SerializedSubject 序列化Subject为线程安全的Subject
*/
public RxBus() {
bus = new SerializedSubject<>(PublishSubject.create());
}
然后我们来发布
public void post(int code, Object obj) {
bus.onNext(obj);
}
最后订阅
public <T> Observable tObservable(final Class<T> eventType) {
return bus.ofType(eventType.getClass());//判断接收事件类型
}
使用
APIServiceManager.getInstance()
.getTravelNotesAPI()
//Retrofit发送请求并返回数据
.getTravelNotesList(query, page + "")
//生命周期处理
.compose(bindToLifecycle())
//img01
.compose(RxSchedulersHelper.io_main())
.subscribe(s -> {
//发布
RxBus.getInstance().post(s.getBookses());
});
//订阅
Subscription subscription=RxBus.getInstance().tObservable(POJO.class)
.subscribe(s -> {//事件处理});
//取消订阅
subscription.unsubscribe();
img01
现在最原始的rxbus就成功了，现在可以进行简单的事件的发布与订阅，虽然可以进行发布于订阅，但是在项目中还没有办法满足我们的需求，那么下面我们就需要对现在的这种方式进行优化。
2.2 rxbus优化
先看一下优化后rxbus包的内容
Paste_Image.png
Subscribe //订阅注解
EventTag //事件标示
EventThread //事件线程
Msg //发布事件时默认封装pojo
RxBus //rxbus主要处理
一共五个类主要的就是rxbus，那我们就先看一下rxbus中到底做了哪些处理
2.2.1 注册
然后需要一个map空间存放订阅者方便取消注册
//存放订阅者信息
private Map<Object, List<Subscription>> subscriptions = new HashMap<>();
现在来处理注册事件
下面代码核心思想是，获取Object 中被Subscribe注解的方法并通过反射的方式，在数据被发布后通过invoke调用方法，并在订阅后把订阅者加入map空间方便解除注册。
public void register(Object subscriber) {
Observable.just(subscriber)
//判断订阅者不为空
.filter(s -> s != null)
.map(s -> s.getClass())
//获取订阅者方法并且用Observable装载
.flatMap(s -> Observable.from(s.getDeclaredMethods()))
//方法必须被Subscribe注解
.filter(m -> m.isAnnotationPresent(Subscribe.class))
//使非public方法可以被invoke
.doOnNext(m -> m.setAccessible(true))
.subscribe(m -> {
addSubscription(m,subscriber);
});
}
private void addSubscription(Method m,Object subscriber){
//获取方法内参数
Class[] parameterType = m.getParameterTypes();
//只获取第一个方法参数，否则默认为Object
Class cla = Object.class;
if (parameterType.length > 1) {
cla = parameterType[0];
}
//获取注解
Subscribe sub = m.getAnnotation(Subscribe.class);
//订阅事件
Subscription subscription = tObservable(sub.tag(), cla)
.observeOn(EventThread.getScheduler(sub.thread()))
.subscribe(o -> {
try {
m.invoke(subscriber, o);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
}, e -> System.out.println("this object is not invoke"));
putSubscriptionsData(subscriber,subscription);
}
/**
* 添加订阅者到map空间来unRegister
* @param subscriber 订阅者
* @param subscription 订阅者 Subscription
*/
private void putSubscriptionsData(Object subscriber,Subscription subscription){
List<Subscription> subs = subscriptions.get(subscriber);
if (subs == null) {
subs = new ArrayList<>();
}
subs.add(subscription);
subscriptions.put(subscriber, subs);
}
代码中我们调用过这么一个方法：tObservable(sub.tag(), cla)
这个就是刚才的tObservable方法，然后增加了一个int类型的参数用于标示在同一个Object下使用不同的事件处理方法
/**
* 订阅事件
* @return
*/
public <T> Observable tObservable(int code, final Class<T> eventType) {
return bus.ofType(Msg.class)//判断接收事件类型
.filter(new Func1<Msg, Boolean>() {
@Override
public Boolean call(Msg o) {
//过滤code相同的事件
return o.code == code;
}
})
.map(new Func1<Msg, Object>() {
@Override
public Object call(Msg o) {
return o.object;
}
})
.cast(eventType);
}
对于post也相应的增加了int参数，通过对code的判断bus会返回eventType类型的Observable数据，并传入备注解的相应方法中。这样就完成了整个事件的注册过程
2.2.2 发布数据
把数据包装以下并发送
public void post(int code, Object obj) {
bus.onNext(new Msg(code, obj));
}
2.2.3 取消注册
public void unRegister(Object subscriber) {
if (subscriber == null) {
return;
}
List<Subscription> subs = subscriptions.get(subscriber);
if (subs != null) {
for (Subscription sub : subs) {
if (sub != null)
sub.unsubscribe();
}
subscriptions.remove(subscriber);
}
}
取消注册主要就是判断map中相同的subscriber，然后调用unsubscribe()取消事件订阅并删除相应的map数据，保证订阅事件与activity生命周期相同防止内存溢出。
使用
APIServiceManager.getInstance()
.getTravelNotesAPI()
.getTravelNotesList(query, page + "")
.compose(bindToLifecycle())
.compose(RxSchedulersHelper.io_main())
//对返回数据的优先处理，如果数据有异常则会执行error
.compose(SchedulersHelper.handleResult())
.subscribe(s -> {
//发布
RxBus.getInstance().post(0, s.getBookses());
},
e -> {
//发布（error）
RxBus.getInstance().post(RxBus.TAG_ERROR, e.getMessage());
});
@Subscribe(tag = 0, thread = EventThread.MAIN_THREAD)
private void dataBinding(ArrayList<TravelNoteBook.Books> bookses) {
//事件处理
}
@Subscribe(tag =RxBus.TAG_ERROR)
private void dataError(String error) {
ToastUtil.getInstance().makeShortToast(this, error);
}
3 总结
现在优化过后最简单的rxbus事件总线就已经实现，可以对数据进行发布和订阅处理，相关的代码和demo可以访问我的git:
https://github.com/hackerlc/GearApplication
或者在gradle中直接引用：
compile 'com.joker.gear:com-joker-gear:1.4.1'
当然因为现在rxbus是最简单的版本，不论是代码还是实现方式还有很多不足的地方，在实际项目中也会遇到更多的问题，这里只提供了只用rxbus的实现思路，随着rxbus的不断完善相信后面会有更加健壮的代码呈现出来。
4 相关引用
在实现rxbus上看了很多的文章和git，也非常感谢这些作者的分享，当然如果在文章引用过程中有什么问题，请联系我我会作出修改，以下是相关的引用连接。
http://www.open-open.com/lib/view/open1444011371541.html
http://www.jianshu.com/p/ca090f6e2fe2
https://github.com/AndroidKnife/RxBus
https://github.com/1030310877/JoeRxBus
https://github.com/ch331917692/RxBus
http://www.jianshu.com/p/f3f0eccbcd6f
