RxBus2（一）讲解

在此前我们要想在activity或fragment之间传递信息，通常的做法是通过startActivityForResult来启动activity，然后在onActvityResult方法中接收信息。这样的做法在少量的代码下是完全可以胜任的，但一旦我们的代码量达到一定的程度时，这样的做法就显的比较复杂繁琐了。页面一多，跳转来跳转去，最终会使的我们自己都要晕头转向了，容易出错。这样的做法也使得代码的后期维护比较困难。这时我们就急需一种简便的方法来替代它，随之应运而生就有EventBus和otto等框架，大家可以去google了解下这些框架，这里就不多讲。这里我们需要详细讲解的是运用RxBus2来替代EventBus中的功能。

在阅读本文章之前请确保你了解或者使用过RxJava/RxAndroid。RxJava出来也有一段时间了，如今也多app都使用了RxJava中的响应式编程。这种编程方式清晰明了，大家也可以去google了解下。

市面上也已经有很多关于使用RxBus来替代EventBus中的功能的文章，之前我也搜索过，大多数都是基于Rxjava1.x版本写的，而Rxjava2.x的改动有点大，有些已经有不适用了。而本文章要讲的RxBus2就是基于Rxjava2.x来编写的。

RxBus2源码：https://github.com/Jasonzhou613/JRxBus2

如何使用：RxBus2（二）如何使用

RxBus2的代码结构很简单，其中最重要的类是RxBus2



首先我们来看register方法

/** 注册 */
public void register(Object subscriber) {
//避免重复
if (subscribersMap.containsKey(subscriber)) {
return;
}
Class<?> subClass = subscriber.getClass();
Method[] methods = subClass.getDeclaredMethods();
for (Method method : methods) {
if (method.isAnnotationPresent(Subscribe.class)) {
//获得参数类型
Class[] parameterType = method.getParameterTypes();
//参数不为空 且参数个数为1
if (parameterType != null && parameterType.length == 1) {
Subscribe sub = method.getAnnotation(Subscribe.class);
int code = sub.code();
ThreadMode threadMode = sub.threadMode();
boolean receiveStickyEvent = sub.receiveStickyEvent();

Class eventType = parameterType[0];
SubscriberMethodInfo subscriberMethodInfo = new SubscriberMethodInfo(subscriber,
method, eventType, code, threadMode, receiveStickyEvent);

addSubscriber(subscriber);
addSubscriberMethodInfoToMap(subscriber, subscriberMethodInfo);
bindObservable(subscriberMethodInfo);
}
}
}

printlnRxBusInfo();
}

此方法会首先判断订阅者是否已经订阅了，如果已经订阅了，则直接返回。如果还未订阅过，则会遍历订阅者所有的方法，将所有有Subscribe修饰的方法并且有且只有一个参数的方法提取并保存起来。

绑定订阅关系

/** 绑定observable */
private void bindObservable(final SubscriberMethodInfo subscriberMethodInfo) {
try {
Disposable disposable = toObservable(Message.class, subscriberMethodInfo)
//.observeOn(Schedulers.io())
//在io线程中通过subscriber和方法参数，筛选出所有符合执行条件的方法，并放回
//这里考虑到，如果项目比大时，注册的方法可能比较多，所以筛选工作放在io线程中，避免引起界面卡顿
.map(new Function<Message, List<InvokeMethodInfo>>() {
@Override
public List<InvokeMethodInfo> apply(Message message) throws Exception {
println("map->apply to List<InvokeMethodInfo>, thread:" + Thread.currentThread());
List<InvokeMethodInfo> methodInfos = getInvokeMethods(subscriberMethodInfo, message);
//如果是粘性事件，计算粘性事件的课消费次数
if (message instanceof StickyMessage) {
StickyMessage stickyMessage = (StickyMessage) message;
int canExecuteTimes = stickyMessage.getCanExecuteTimes();
//只有消费次数大于等于0的粘性事件才会消亡，消费次数小于0的粘性事件是不会消亡的
if (canExecuteTimes >= 0) {
//若粘性事件剩余消费次数小于方法数，则表明粘性事件的消费次数不足，有些方法不能执行到，这里将多出来的方法移除掉
if (canExecuteTimes < methodInfos.size()) {
Class<?> key = subscriberMethodInfo.getEventType();
stickyEventMap.remove(key);
stickyEventKeys.remove(key);
String methodNames = "";
while (methodInfos.size() > canExecuteTimes) {
//移除最后一个
InvokeMethodInfo info = methodInfos.remove(methodInfos.size() - 1);
methodNames = methodNames + "," + info.getSubscriberMethodInfo().getMethod().getName();
}
methodNames = methodNames.replaceFirst(",", "");
println("The " + stickyMessage.toString() + " canExecuteTimes is not enough, the method["
+ methodNames + "] will not invoked.");
} else {
stickyMessage.setCanExecuteTimes(stickyMessage.getCanExecuteTimes() - methodInfos.size());
}
}
printlnStickyEvent();
}

return methodInfos;
}
})
//指定observeOn线程
.observeOn(getObserveOnScheduler(subscriberMethodInfo.getThreadMode()))
.subscribe(
new Consumer<List<InvokeMethodInfo>>() {
@Override
public void accept(List<InvokeMethodInfo> invokeMethodInfos) throws Exception {
println("subscribe->accept, thread:" + Thread.currentThread());
if (invokeMethodInfos == null) {
return;
}
//执行上一步map方法所返回的方法集合
for (int i = 0; i < invokeMethodInfos.size(); i++) {
invokeMethodInfos.get(i).invoke();
}
}
},
new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
println("thread:" + Thread.currentThread());
String msg = "Error msg:";
if (throwable != null) {
msg = msg + throwable.getMessage();
}
JRxBusLog.e(TAG, "throwable, msg:" + msg);
}
});

addDisposableToMap(subscriberMethodInfo.getSubscriber(), disposable);

} catch (Exception e) {
e.printStackTrace();
JRxBusLog.e(TAG, "Exception e:" + e.getMessage());
}
}

根据当前的订阅的方法信息实例和发送的消息筛选出符合执行条件的方法

private List<InvokeMethodInfo> getInvokeMethods(SubscriberMethodInfo currentMethodInfo, Message message) {
List<InvokeMethodInfo> invokeMethodInfos = new ArrayList<>();

for (Map.Entry<Object, List<SubscriberMethodInfo>> entry : subscriberMethodInfosMap.entrySet()) {
List<SubscriberMethodInfo> methodInfos = entry.getValue();
//跳过空值
if (methodInfos == null) {
continue;
}
for (int i = 0; i < methodInfos.size(); i++) {
SubscriberMethodInfo methodInfo = methodInfos.get(i);
Class objectType = message.getObject().getClass();
//订阅方法中的订阅者与当前订阅值是同一个，并且订阅事件类型一样，则符合条件
if (currentMethodInfo == methodInfo
&& objectType == methodInfo.getEventType()
&& message.getCode() == methodInfo.getCode()) {

InvokeMethodInfo invokeMethodInfo = new InvokeMethodInfo(methodInfo, message.getObject());

//如果是粘性事件，则还需判断该方法是否声明了接收粘性事件
if (message instanceof StickyMessage) {
if (methodInfo.isReceiveStickyEvent()) {
invokeMethodInfos.add(invokeMethodInfo);
}

} else {
if (!methodInfo.isReceiveStickyEvent()) {
invokeMethodInfos.add(invokeMethodInfo);
}
}
}
}
}

return invokeMethodInfos;
}

在Activity或者Fragment onDestroy的时候，我们需要取消订阅关系

[java] view plain copy
/** 反注册 */
public void unRegister(Object subscriber) {
unSubscribeBySubscriber(subscriber);
subscriberMethodInfosMap.remove(subscriber);
subscribersMap.remove(subscriber);

printlnRxBusInfo();
}

/**
* 取消订阅关系
*
* @param subscriber key
*/
private void unSubscribeBySubscriber(Object subscriber) {
List<Disposable> disposables = disposablesMap.get(subscriber);
if (disposables != null) {
while (disposables.size() > 0) {
Disposable disposable = disposables.remove(0);
if (disposable != null) {
disposable.dispose();
}
}
}
disposablesMap.remove(subscriber);
}