If you just a little want,you will only get a little.
1# 概述
Android 社区有名的事件总线库有EventBus，otto。后者为支持RxJava而主动Deprecated。可以看得出Otto的创建者非常大度，另一方面也说明了RxJava能更好地实现事件总线。(JakeWharton也是RxJava的主要贡献者)。
为减少读者学习成本，本文高仿EventBus来打造一个基于RxJava2的小型事件总线RxBus2。如果在过程中大家看到EventBus的影子不必惊慌，说明你对EventBus吃得比较透。
通过本文可以GET
Flowable的创建，过滤，线程调度，订阅，事件发射，Flowable的取消，自定义注解，反射获取方法信息，反射调用方法等知识点的肤浅认识。如果不感兴趣可以先走一步。
2# 基本操作
既然是高仿，RxBus2的API命名方式也得和EventBus保持一致。基本操作如下：
- 注册
- 注解方式处理事件
- 发射事件
- 取消注册
首先以单例模式提供RxBus，构造方法内创建了FlowableProcessor和SubscriberMethodFinder。FlowableProcessor(来自RxJava2)对应RxJava中的PublishSubject，是Flowable版的Subject，仅会向Subsciber发射在订阅之后Flowable发射的数据。SubscriberMethodFinder来自EventBus，用于解析注解方法，具体使用后面会讲到。
private RxBus() {
mFlowableProcessor = PublishProcessor.create().toSerialized();
mSubscriberMethodFinder = new SubscriberMethodFinder();
}
public static RxBus getDefault() {
if (mRxBus == null) {
synchronized (RxBus.class) {
if (mRxBus == null) {
mRxBus = new RxBus();
}
}
}
return mRxBus;
}
2.1# 注册
public void register(Object subsciber) {
Class<?> subsciberClass = subsciber.getClass();
List<SubscriberMethod> subscriberMethods = mSubscriberMethodFinder.findSubscriberMethods(subsciberClass);
for (SubscriberMethod subscriberMethod : subscriberMethods) {
addSubscriber(subsciber, subscriberMethod);
}
}
采用反射方式获取Subscribe注解方法的信息，包括Method对象，参数类型，线程模式。阅读过EventBus源码的同学可以跳过，避免产生不屑之情。
public List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
List<SubscriberMethod> subscriberMethods = findUsingReflection(subscriberClass);
if (subscriberMethods.isEmpty()) {
throw new RxBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
}
return subscriberMethods;
}
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
List<SubscriberMethod> subscriberMethods = new ArrayList<>();
Method[] methods = subscriberClass.getDeclaredMethods();
for (Method method : methods) {
int modifiers = method.getModifiers();
//@Subscribe 的注解方法必须是 public, non-static, and non-abstract
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
Class<?>[] parameterTypes = method.getParameterTypes();
//@Subscribe 的注解方法只能且必须有一个参数
if (parameterTypes.length == 1) {
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
//必须有@Subscribe 注解
if (subscribeAnnotation != null) {
ThreadMode threadMode = subscribeAnnotation.threadMode();
Class<?> eventType = parameterTypes[0];
subscriberMethods.add(new SubscriberMethod(method, eventType, getThreadMode(threadMode)));
}
}
}
}
return subscriberMethods;
}
完成Subscribe注解方法解析后，根据方法信息生成订阅者，并存储返回的Disposable，以便以后取消注册用。
private void addSubscriber(final Object subsciber, final SubscriberMethod subscriberMethod) {
Class<?> subsciberClass = subsciber.getClass();
Class<?> eventType = subscriberMethod.getEventType();
//ofType事件过滤，observeOn线程调度，subscribe订阅回调
Disposable disposable = mFlowableProcessor.ofType(eventType).observeOn(subscriberMethod.getThreadMode())
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
invokeMethod(subsciber, subscriberMethod, o);//反射调用注解方法
}
});
//Map<SubscriberClass,Map<EventTypeClass,Disposable>>存储Disposable作取消事件之用
Map<Class<?>, Disposable> disposableMap = mDisposableMap.get(subsciberClass);
if (disposableMap == null) {
disposableMap = new HashMap<>();
mDisposableMap.put(subsciberClass, disposableMap);
}
disposableMap.put(eventType, disposable);
}
反射调用注解方法
private void invokeMethod(Object subscriber, SubscriberMethod subscriberMethod, Object obj) {
try {
subscriberMethod.getMethod().invoke(subscriber, obj);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
}
2.2# 发射事件
不管有没订阅者都进行发射，以备扩展Sticky事件(目前没实现)
public void post(Object obj) {
mFlowableProcessor.onNext(obj);
}
2.3# 取消注册
取消注册有两种，原理是取出存储在Map中的 Disposable进行dispose()，然后remove。
- 根据订阅者取消(取消订阅者中所有的订阅方法)
public void unregister(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
Map<Class<?>, Disposable> disposableMap = mDisposableMap.get(subscriberClass);
if (disposableMap == null) {
throw new IllegalArgumentException(subscriberClass.getSimpleName() + " haven't registered RxBus");
}
Set<Class<?>> keySet = disposableMap.keySet();
for (Class<?> evenType : keySet) {
Disposable disposable = disposableMap.get(evenType);
disposable.dispose();
}
mDisposableMap.remove(subscriberClass);
}
- 根据订阅事件类型取消(仅取消订阅者特定事件类型的订阅方法)
public void unregister(Object subscriber, Class<?> eventType) {
Class<?> subscriberClass = subscriber.getClass();
Map<Class<?>, Disposable> disposableMap = mDisposableMap.get(subscriberClass);
if (disposableMap == null) {
throw new IllegalArgumentException(subscriberClass.getSimpleName() + " haven't registered RxBus");
}
if (!disposableMap.containsKey(eventType)) {
throw new IllegalArgumentException("The event with type of " + subscriberClass.getSimpleName() + " is not" +
" required in " + subscriberClass.getSimpleName());
}
Disposable disposable = disposableMap.get(eventType);
disposable.dispose();
mDisposableMap.remove(eventType);
}
总结
基于RxJava打造RxBus可谓物尽其用。RxJava管理着可观测，可控制的事件序列，比起EventBus有更多得天独厚的优势，它的线程调度，各种类型的操作符，使得在实际运用中可以在事件控制方面进行大量的自定义操作。
对本文源码有兴趣的同学可以在此传送【RxBus2】
