手写EventBus框架——动手_终结
2017-03-26 本文已影响625人
wenld_
路漫漫其修远兮
01. 手写EventBus框架——源码分析1
02. 手写EventBus框架——源码分析2
03. 手写EventBus框架——动手_整体架构设计
04. 手写EventBus框架——动手_终结
上一篇文章已经把我们的整体架构已经搭建好了，实现原理和技术在我们分析源码的时候也已经知道，现在我们来实现它吧。
先实现我们的工具类；
SubscriberMethodFinder.class
/**
* 订阅方法查找类
* <p/>
* Author: 温利东 on 2017/3/26 11:21.
* blog: http://www.jianshu.com/u/99f514ea81b3
* github: https://github.com/LidongWen
*/
public class SubscriberMethodFinder {
private static final int MODIFIERS_IGNORE = Modifier.ABSTRACT | Modifier.STATIC;
private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
/**
* 查找方法
*
* @param subscriberClass
* @return
*/
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
subscriberMethods = findUsingReflection(subscriberClass);
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
List<SubscriberMethod> subscriberMethods = new ArrayList<>();
Class<?> clazz = subscriberClass;
boolean skipSuperClasses = false;
while (clazz != null) {
Method[] methods;
try {
methods = subscriberClass.getDeclaredMethods();
} catch (Throwable th) {
methods = subscriberClass.getMethods();
skipSuperClasses = true;
}
for (Method method : methods) {
int modifiers = method.getModifiers();
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 1) {
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
if (subscribeAnnotation != null) {
Class<?> eventType = parameterTypes[0];
ThreadMode threadMode = subscribeAnnotation.threadMode();
subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
}
} else if (method.isAnnotationPresent(Subscribe.class)) {
String methodName = method.getDeclaringClass().getName() + "." + method.getName();
throw new EventBusException("@Subscribe method " + methodName +
"must have exactly 1 parameter but has " + parameterTypes.length);
}
} else if (method.isAnnotationPresent(Subscribe.class)) {
String methodName = method.getDeclaringClass().getName() + "." + method.getName();
throw new EventBusException(methodName +
" is a illegal @Subscribe method: must be public, non-static, and non-abstract");
}
}
if (skipSuperClasses) {
clazz = null;
} else {
clazz = clazz.getSuperclass();
String clazzName = clazz.getName();
/** Skip system classes, this just degrades performance. */
if (clazzName.startsWith("java.") || clazzName.startsWith("javax.") || clazzName.startsWith("android.")) {
clazz = null;
}
}
}
return subscriberMethods;
}
}
InvokeHelper.class
/**
* 方法执行帮助类
* <p/>
* Author: 温利东 on 2017/3/26 11:29.
* blog: http://www.jianshu.com/u/99f514ea81b3
* github: https://github.com/LidongWen
*/
final class InvokeHelper {
private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();
private static InvokeHelper ourInstance;
private HandlerPoster handlerPoster;
public static InvokeHelper getDefault() {
if (ourInstance == null) {
synchronized (InvokeHelper.class) {
if (ourInstance == null) {
ourInstance = new InvokeHelper();
}
}
}
return ourInstance;
}
private InvokeHelper() {
handlerPoster = new HandlerPoster(Looper.getMainLooper());
}
public void post(final Subscription subscription, final Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case POSTING:
//直接执行
invokeSubscriber(subscription, event);
break;
case MAIN:
if (isMainThread) {
//直接执行
invokeSubscriber(subscription, event);
} else {
// 放在handler内执行
handlerPoster.post(new Runnable() {
@Override
public void run() {
invokeSubscriber(subscription, event);
}
});
}
break;
case BACKGROUND:
if (isMainThread) {
//放在后台线程执行
// getExecutorService().execute(new Runnable() {
// @Override
// public void run() {
// invokeSubscriber(subscription,event);
// }
// });
} else {
//执行
invokeSubscriber(subscription, event);
}
break;
case ASYNC:
//放在异步线程内执行
getExecutorService().execute(new Runnable() {
@Override
public void run() {
invokeSubscriber(subscription, event);
}
});
break;
default:
//抛异常
throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
}
}
private void invokeSubscriber(Subscription subscription, Object event) {
try {
subscription.subscriberMethod.method.invoke(subscription.subscriberMethod.method, event);
} catch (InvocationTargetException e) {
// throw new InvocationTargetException(subscriberMethod.subscriber,e.getCause(),event);
} catch (IllegalAccessException e) {
throw new IllegalStateException("Unexpected exception", e);
}
}
ExecutorService getExecutorService() {
return DEFAULT_EXECUTOR_SERVICE;
}
class HandlerPoster extends Handler {
HandlerPoster(Looper looper) {
super(looper);
}
}
}
EventBus.class
/**
* <p/>
* Author: 温利东 on 2017/3/26 10:44.
* blog: http://www.jianshu.com/u/99f514ea81b3
* github: https://github.com/LidongWen
*/
public class EventBus {
public static String TAG = "EventBus";
private static EventBus ourInstance;
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
private final Map<Object, List<Class<?>>> typesBySubscriber;
//粘性
private final Map<Class<?>, Object> stickyEvents;
private final SubscriberMethodFinder subscriberMethodFinder;
//每个线程内的一个队列
private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
};
public static EventBus getDefault() {
if (ourInstance == null) {
synchronized (EventBus.class) {
if (ourInstance == null) {
ourInstance = new EventBus();
}
}
}
return ourInstance;
}
private EventBus() {
//初始化一些数据
subscriptionsByEventType = new HashMap<>();
stickyEvents = new ConcurrentHashMap<>();
typesBySubscriber = new HashMap<>();
subscriberMethodFinder = new SubscriberMethodFinder();
}
public void register(Object subscriber) {
//1.找到所有方法
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
//2.保存订阅 subscriber();
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
/**
* 保存订阅方法
*
* @param subscriber
*/
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//1.保存数据 ， 如果重复 抛异常
Class<?> eventType = subscriberMethod.eventType;
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
subscriptionsByEventType.put(eventType, subscriptions);
} else {
if (subscriptions.contains(newSubscription)) {
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
}
int size = subscriptions.size();
for (int i = 0; i <= size; i++) {
if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
subscriptions.add(i, newSubscription);
break;
}
}
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//2.执行粘性事件
if (subscriberMethod.sticky) {
// Existing sticky events of all subclasses of eventType have to be considered.
// Note: Iterating over all events may be inefficient with lots of sticky events,
// thus data structure should be changed to allow a more efficient lookup
// (e.g. an additional map storing sub classes of super classes: Class -> List<Class>).
Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
for (Map.Entry<Class<?>, Object> entry : entries) {
Class<?> candidateEventType = entry.getKey();
if (eventType.isAssignableFrom(candidateEventType)) {
Object stickyEvent = entry.getValue();
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
}
}
private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
if (stickyEvent != null) {
// If the subscriber is trying to abort the event, it will fail (event is not tracked in posting state)
// --> Strange corner case, which we don't take care of here.
postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
}
}
private void postToSubscription(Subscription newSubscription, Object stickyEvent, boolean b) {
InvokeHelper.getDefault().post(newSubscription, stickyEvent, b);
}
public synchronized void unRegister(Object subscriber) {
//去除订阅
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
if (subscribedTypes != null) {
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);
}
typesBySubscriber.remove(subscriber);
} else {
Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions != null) {
int size = subscriptions.size();
for (int i = 0; i < size; i++) {
Subscription subscription = subscriptions.get(i);
if (subscription.subscriber == subscriber) {
subscription.active = false;
subscriptions.remove(i);
i--;
size--;
}
}
}
}
public void post(Object event) {
//1.放入执行队列
PostingThreadState postingState = currentPostingThreadState.get();
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);
if (!postingState.isPosting) {
postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
postingState.isPosting = true;
if (postingState.canceled) {
throw new EventBusException("Internal error. Abort state was not reset");
}
try {
while (!eventQueue.isEmpty()) {
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
subscriptionFound |= postSingleEventForEventType(event, postingState, eventClass);
if (!subscriptionFound) {
}
}
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
CopyOnWriteArrayList<Subscription> subscriptions;
synchronized (this) {
subscriptions = subscriptionsByEventType.get(eventClass);
}
if (subscriptions != null && !subscriptions.isEmpty()) {
for (Subscription subscription : subscriptions) {
postingState.event = event;
postingState.subscription = subscription;
boolean aborted = false;
try {
postToSubscription(subscription, event, postingState.isMainThread);
aborted = postingState.canceled;
} finally {
postingState.event = null;
postingState.subscription = null;
postingState.canceled = false;
}
if (aborted) {
break;
}
}
return true;
}
return false;
}
public void postSticky(Object event) {
//加入粘性缓存 stickyEvents
synchronized (stickyEvents) {
stickyEvents.put(event.getClass(), event);
}
//执行
post(event);
}
public void cancelEventDelivery(Object event) {
PostingThreadState postingState = currentPostingThreadState.get();
if (!postingState.isPosting) {
throw new EventBusException(
"This method may only be called from inside event handling methods on the posting thread");
} else if (event == null) {
throw new EventBusException("Event may not be null");
} else if (postingState.event != event) {
throw new EventBusException("Only the currently handled event may be aborted");
} else if (postingState.subscription.subscriberMethod.threadMode != ThreadMode.POSTING) {
throw new EventBusException(" event handlers may only abort the incoming event");
}
postingState.canceled = true;
}
/**
* Gets the most recent sticky event for the given type.
*
* @see #postSticky(Object)
*/
public <T> T getStickyEvent(Class<T> eventType) {
synchronized (stickyEvents) {
return eventType.cast(stickyEvents.get(eventType));
}
}
/**
* Remove and gets the recent sticky event for the given event type.
*
* @see #postSticky(Object)
*/
public <T> T removeStickyEvent(Class<T> eventType) {
synchronized (stickyEvents) {
return eventType.cast(stickyEvents.remove(eventType));
}
}
/**
* Removes the sticky event if it equals to the given event.
*
* @return true if the events matched and the sticky event was removed.
*/
public boolean removeStickyEvent(Object event) {
synchronized (stickyEvents) {
Class<?> eventType = event.getClass();
Object existingEvent = stickyEvents.get(eventType);
if (event.equals(existingEvent)) {
stickyEvents.remove(eventType);
return true;
} else {
return false;
}
}
}
/**
* Removes all sticky events.
*/
public void removeAllStickyEvents() {
synchronized (stickyEvents) {
stickyEvents.clear();
}
}
final static class PostingThreadState {
final List<Object> eventQueue = new ArrayList<Object>();
boolean isPosting;
boolean isMainThread;
Subscription subscription;
Object event;
boolean canceled;
}
}
这边我们就实现一个简易版的EventBus了。
** 感受 **
不得不说，这套框架非常优秀
并发性健壮性来讲：用到了同步锁，队列设计，CopyOnWriteArrayList，ConcurrentHashMap等用来保证线程安全 ；
设计来讲：非常巧妙，各种数据缓存,较少了数据检索。
整体架构来讲：牛！
** 自愧还差得远呢。。。o.O 还要加倍努力！！！**
github：https://github.com/LidongWen/EventBusWenld
希望我的文章不会误导在观看的你，如果有异议的地方欢迎讨论和指正。
如果能给观看的你带来收获，那就是最好不过了。
