Android--源码分析EventBus及手写实现

EventBus为第三方消息通信的框架，因为使用比Handler便利，广受开发者喜爱，其底层实现还是利用的Handler，在其基础上增加了注解，并根据注解在内部实现线程切换接收消息

EventBus使用只有简单的三步：

• 注解方法
• 调用register方法注册
• 调用post方法发送消息

一、EventBus源码分析

源码主要从两个方法入手，就是register方法和post方法

获取EventBus实例的getDefault方法就是一个单例：

    public static EventBus getDefault() {
        if (defaultInstance == null) {
            synchronized (EventBus.class) {
                if (defaultInstance == null) {
                    defaultInstance = new EventBus();
                }
            }
        }
        return defaultInstance;
    }

此外，它还支持使用EventBusBuilder自定义构建实例，感兴趣的自己查看下源码

1.register

首先来看register方法的具体实现：

    public void register(Object subscriber) {
        Class<?> subscriberClass = subscriber.getClass();
        // 获取被注解的方法
        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
        synchronized (this) {
            // 订阅
            for (SubscriberMethod subscriberMethod : subscriberMethods) {
                subscribe(subscriber, subscriberMethod);
            }
        }
    }

其中会获取注册对象的类对象，并调用findSubscriberMethods方法获取类中EventBus注解的方法，并将方法包装成SubscriberMethod列表，存入以注册对象的Class为key的METHOD_CACHE 这个Map中，findSubscriberMethods方法的调用链实现如下：

    // (Class : 方法包装类列表)的缓存
    private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();

    // 获取类中EventBus注解的方法
    List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
        List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
        if (subscriberMethods != null) {// 获取到缓存，直接返回
            return subscriberMethods;
        }

        if (ignoreGeneratedIndex) {
            // 利用反射获取注解方法列表
            subscriberMethods = findUsingReflection(subscriberClass);
        } else {
            subscriberMethods = findUsingInfo(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) {
        FindState findState = prepareFindState();
        findState.initForSubscriber(subscriberClass);
        while (findState.clazz != null) {
            // 利用反射获取当前类的方法
            findUsingReflectionInSingleClass(findState);
            // 向上查询父类中被注解的方法
            findState.moveToSuperclass();
        }
        // 将方法列表返回
        return getMethodsAndRelease(findState);
    }

    // 真正利用反射获取的方法
    private void findUsingReflectionInSingleClass(FindState findState) {
        Method[] methods;
        try {
            // This is faster than getMethods, especially when subscribers are fat classes like Activities
            // 获取当前类的所有方法
            methods = findState.clazz.getDeclaredMethods();
        } catch (Throwable th) {
            // Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149
            methods = findState.clazz.getMethods();
            findState.skipSuperClasses = true;
        }
        for (Method method : methods) {
            int modifiers = method.getModifiers();
            // 如果是Public方法，并且不是抽象、静态等方法
            if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
                Class<?>[] parameterTypes = method.getParameterTypes();
                if (parameterTypes.length == 1) {// EventBus只支持一个参数
                    // 获取Subscribe注解
                    Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
                    if (subscribeAnnotation != null) {
                        // 获取入参的类对象
                        Class<?> eventType = parameterTypes[0];
                        if (findState.checkAdd(method, eventType)) {// 判断下有没有添加过
                            ThreadMode threadMode = subscribeAnnotation.threadMode();
                            // 包装后，添加到方法列表中
                            findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                    subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                        }
                    }
                } else if (strictMethodVerification && 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 (strictMethodVerification && 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");
            }
        }
    }

回到register方法中，最后调用了subscribe方法，入参为注册对象和包装方法类SubscriberMethod，其中又做了缓存：

• 将注册对象和方法包装类SubscriberMethod重新包装成Subscription对象，并将入参类型为Key，包装类Subscription列表为Value存入subscriptionsByEventType这个Map中
• 再将该订阅方法的参数类型存入以注册对象为key，以参数类型集合为value的typesBySubscriber这个Map中

    private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
        // 入参参数的类对象
        Class<?> eventType = subscriberMethod.eventType;
        // 将方法包装类对象subscriberMethod和注册时传入对象包装为Subscription类
        Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
        CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
        if (subscriptions == null) {
            subscriptions = new CopyOnWriteArrayList<>();
            // 以入参参数的类对象为Key，Subscription列表为Value进行缓存
            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++) {// 遍历包装类Subscription集合，根据优先级插入新包装的Subscription
            if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
                subscriptions.add(i, newSubscription);
                break;
            }
        }

        // 获取参数类型集合（EventBus的订阅方法为：一个方法对应一个参数类型）
        List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
        if (subscribedEvents == null) {
            subscribedEvents = new ArrayList<>();
            // 存入以注册对象为key，参数类型集合为value的Map中
            typesBySubscriber.put(subscriber, subscribedEvents);
        }
        // 参数类型列表添加该订阅方法的参数类型
        subscribedEvents.add(eventType);

        if (subscriberMethod.sticky) {// 默认为false，暂不做研究
            if (eventInheritance) {
                // 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);
                    }
                }
            } else {
                Object stickyEvent = stickyEvents.get(eventType);
                checkPostStickyEventToSubscription(newSubscription, stickyEvent);
            }
        }
    }

总的来说，register方法做的事情就是利用反射获取方法，往三个Map中存入对应的值，再来看下这三个Map:

• METHOD_CACHE ：以注册对象的Class为key，以包装方法类SubscriberMethod列表为value

• subscriptionsByEventType：以订阅方法的唯一入参参数：Class为key，以Subscription（包装了注册对象和SubscriberMethod（包装了Method、注解信息等））列表为value

• typesBySubscriber：以注册对象为key，以订阅方法的参数类型Class集合为value

简单分析下这三个Map的作用：

• 当发送消息时，可以通过subscriptionsByEventType快速获取到Subscription列表，进而进行方法的反射调用，反射需要对象和Method，这两个Subscription都有

• 当调用unregister方法，取消订阅时，通过typesBySubscriber可以快速获取到订阅方法的参数类型Class集合，通过遍历参数类型Class集合，移除subscriptionsByEventType中以这些Class为key的Subscription集合，这样注册时的对象就不会内存泄漏了

• 至于METHOD_CACHE，存放着Class和Method的包装类SubscriberMethod，以便通过Class快速获取这个类的SubscriberMethod（Method、注解信息等）的列表，本来通过Class就可以获取到Method集合，而Class加载后一般不会进行垃圾回收，所以不存在内存泄漏问题，必要时也可以通过clear方法清空这个Map

2.post

post方法的调用分析比较简单，这边直接跳到调用链后面的方法：postSingleEventForEventType，该方法根据订阅方法入参的参数类型获取到所有需要通知的方法，最后遍历调用postToSubscription方法进行线程调度：

    // 根据post传入的对象，进行一对多消息分发
    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;
    }

postToSubscription方法会根据注解信息，来进行线程调度，invokeSubscriber方法则会直接利用反射调用订阅方法，而mainThreadPoster对象就是Handler，发送消息后，最终通过Handler的handleMessage方法中在主线程调用invokeSubscriber方法，backgroundPoster、asyncPoster则会通过线程池，在子线程中异步执行

    // 线程调度
    private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
        switch (subscription.subscriberMethod.threadMode) {
            case POSTING:
                invokeSubscriber(subscription, event);
                break;
            case MAIN:
                if (isMainThread) {
                    invokeSubscriber(subscription, event);
                } else {
                    mainThreadPoster.enqueue(subscription, event);
                }
                break;
            case BACKGROUND:
                if (isMainThread) {
                    backgroundPoster.enqueue(subscription, event);
                } else {
                    invokeSubscriber(subscription, event);
                }
                break;
            case ASYNC:
                asyncPoster.enqueue(subscription, event);
                break;
            default:
                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
        }
    }

    // 反射调用方法
    void invokeSubscriber(Subscription subscription, Object event) {
        try {
            subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
        } catch (InvocationTargetException e) {
            handleSubscriberException(subscription, event, e.getCause());
        } catch (IllegalAccessException e) {
            throw new IllegalStateException("Unexpected exception", e);
        }
    }

EventBus的核心设计模式是订阅者模式，利用注解获取到一个对象中所有的订阅方法并缓存，最后利用反射调用这些方法

二、手写实现EventBus

1.定义注解和线程模式

注解：

/**
 * 该注解表示方法需要被EventBus订阅
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Subscribe {
    ThreadMode threadMode() default ThreadMode.POST;
}

线程模式：

public enum ThreadMode {
    Main,//主线程执行订阅方法
    POST,//默认
    BACKGROUND//后台线程订阅执行
}

2.定义方法包装类：SubscriberMethod

/**
 * 方法包装类
 */
class SubscriberMethod {
    final Method method;//方法
    final ThreadMode threadMode;//线程模式
    final Class<?> eventType;//方法的入参参数类型

    public SubscriberMethod(Method method, ThreadMode threadMode, Class<?> eventType) {
        this.method = method;
        this.threadMode = threadMode;
        this.eventType = eventType;
    }
}

3.定义对象 方法包装类：Subscription

/**
 * 对象 方法包装类
 */
class Subscription {
    // 注册对象
    final Object subscriber;
    // 方法包装类
    final SubscriberMethod subscriberMethod;

    public Subscription(Object subscriber, SubscriberMethod subscriberMethod) {
        this.subscriber = subscriber;
        this.subscriberMethod = subscriberMethod;
    }
}

4.定义EventBus类实现功能

public class EventBus {
    private static final int BRIDGE = 0x40;
    private static final int SYNTHETIC = 0x1000;
    private static final int MODIFIERS_IGNORE = Modifier.ABSTRACT | Modifier.STATIC | BRIDGE | SYNTHETIC;

    //以注册对象的Class为key，以包装方法类SubscriberMethod列表为value
    private final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
    //以订阅方法的唯一入参参数：Class为key，以Subscription（包装了注册对象和SubscriberMethod（包装了Method、注解信息等））列表为value
    private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType = new HashMap<>();
    //以注册对象为key，以订阅方法的参数类型Class集合为value
    private final Map<Object, List<Class<?>>> typesBySubscriber = new HashMap<>();
    //线程池
    private final ExecutorService executorService;

    private static final EventBus instance = new EventBus();

    public static EventBus getInstance() {
        return instance;
    }

    private EventBus() {
        executorService = Executors.newCachedThreadPool();
    }

    /**
     * 订阅
     *
     * @param obj
     */
    public void register(Object obj) {
        Class<?> clz = obj.getClass();
        //查找订阅方法
        List<SubscriberMethod> methods = findSubscriberMethod(clz);

        //加入订阅map
        if (!methods.isEmpty()) {
            synchronized (this) {
                subscribe(obj, methods);
            }
        }
    }

    /**
     * 将订阅方法加入两个map缓存
     *
     * @param obj
     * @param methods
     */
    private void subscribe(Object obj, List<SubscriberMethod> methods) {
        //对象中订阅方法的入参类型集合
        Set<Class<?>> ObjEventTypes = new HashSet<>();

        for (SubscriberMethod subscriberMethod : methods) {
            //根据入参类型获取Subscription列表
            Class<?> eventType = subscriberMethod.eventType;
            CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);


            if (subscriptions == null) {
                subscriptions = new CopyOnWriteArrayList<>();
                subscriptionsByEventType.put(eventType, subscriptions);
            }

            //包装成Subscription存入
            subscriptions.add(new Subscription(obj, subscriberMethod));

            //记录入参类型
            ObjEventTypes.add(eventType);
        }


        // 将对象：入参类型列表存入map
        if (typesBySubscriber.get(obj) == null) {
            List<Class<?>> eventTypes = new ArrayList<>(ObjEventTypes);

            typesBySubscriber.put(obj, eventTypes);
        }
    }

    /**
     * 根据class查找订阅方法
     *
     * @param clz
     * @return
     */
    private List<SubscriberMethod> findSubscriberMethod(Class<?> clz) {
        List<SubscriberMethod> methods = METHOD_CACHE.get(clz);
        if (methods != null) {
            return methods;
        }

        // 利用反射获取
        return getMethodsByReflect(clz);
    }

    /**
     * 根据class反射获取订阅方法
     *
     * @param clz
     * @return
     */
    private List<SubscriberMethod> getMethodsByReflect(Class<?> clz) {
        // 订阅方法集合
        List<SubscriberMethod> subscriberMethods = new ArrayList<>();

        while (clz != null && clz != Object.class) {
            Method[] declaredMethods = clz.getDeclaredMethods();

            for (Method method : declaredMethods) {
                Subscribe annotation = method.getAnnotation(Subscribe.class);
                if (!Modifier.isPublic(method.getModifiers()) ||
                        (method.getModifiers() & MODIFIERS_IGNORE) != 0 || annotation == null)
                    continue;

                //获取线程模式
                ThreadMode threadMode = annotation.threadMode();
                //获取入参参数类型
                Class<?>[] parameterTypes = method.getParameterTypes();
                if (parameterTypes.length != 1) throw new RuntimeException("入参参数必须为一个");
                Class<?> eventType = parameterTypes[0];

                // 构造方法包装类
                SubscriberMethod subscriberMethod = new SubscriberMethod(method, threadMode, eventType);

                //加入缓存
                subscriberMethods.add(subscriberMethod);
            }

            clz = clz.getSuperclass();
        }

        if (subscriberMethods.isEmpty()) throw new RuntimeException("该类及其父类没有任何没有订阅方法");

        //加入缓存
        METHOD_CACHE.put(clz, subscriberMethods);
        return subscriberMethods;
    }

    /**
     * 取消订阅
     *
     * @param obj
     */
    public synchronized void unregister(Object obj) {
        //获取该对象订阅方法的所有参数类型
        List<Class<?>> eventTypes = typesBySubscriber.get(obj);
        if (eventTypes != null) {
            for (Class<?> eventType : eventTypes) {
                //获取Subscription集合
                CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
                for (int i = 0; i < subscriptions.size(); i++) {
                    if (subscriptions.get(i).subscriber == obj) {//判断是不是该对象的方法
                        subscriptions.remove(i);//移除
                        i--;
                    }
                }
            }

            // 移除
            typesBySubscriber.remove(obj);
        }
    }

    /**
     * 发送消息
     *
     * @param event
     */
    public void post(Object event) {
        //获取到订阅方法
        CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(event.getClass());
        if (subscriptions != null && !subscriptions.isEmpty()) {
            for (Subscription subscription : subscriptions) {
                postToSubscription(event, subscription.subscriber, subscription.subscriberMethod);
            }
        }
    }

    /**
     * 线程调度
     *
     * @param event
     * @param subscriberMethod
     */
    private void postToSubscription(Object event, Object subscriber, SubscriberMethod subscriberMethod) {
        // 当前线程是否是主线程
        boolean isMainThread = Looper.myLooper() == Looper.getMainLooper();
        switch (subscriberMethod.threadMode) {
            case Main:
                if (isMainThread) {//主线程直接调用
                    invokeSubscriberMethod(event, subscriber, subscriberMethod);
                } else {
                    // 偷懒
                    new Handler(Looper.getMainLooper()).post(new Runnable() {
                        @Override
                        public void run() {
                            invokeSubscriberMethod(event, subscriber, subscriberMethod);
                        }
                    });
                }
                break;
            case POST:
                invokeSubscriberMethod(event, subscriber, subscriberMethod);
                break;
            case BACKGROUND:
                // 偷懒
                getExecutorService().execute(new Runnable() {
                    @Override
                    public void run() {
                        invokeSubscriberMethod(event, subscriber, subscriberMethod);
                    }
                });
                break;
        }
    }

    /**
     * 反射调用订阅方法
     *
     * @param event
     * @param subscriber
     * @param subscriberMethod
     */
    private void invokeSubscriberMethod(Object event, Object subscriber, SubscriberMethod subscriberMethod) {
        try {
            subscriberMethod.method.invoke(subscriber, event);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取线程池
     *
     * @return ExecutorService
     */
    public ExecutorService getExecutorService() {
        return executorService;
    }

}

最后写个按钮，测试下

Activity中的代码如下：

public class MainActivity extends AppCompatActivity {

    @Subscribe
    public void onMessage(String msg) {
        Log.i("aruba", "onMessage:" + msg + "Thread:" + Thread.currentThread().getName());
    }

    @Subscribe(threadMode = ThreadMode.BACKGROUND)
    public void onMessageBackground(String msg) {
        Log.i("aruba", "onMessageBackground:" + msg + "Thread:" + Thread.currentThread().getName());
    }

    public void onMessageNotSubscribe(String msg) {
        Log.i("aruba", "onMessageNotSubscribe:" + msg);
    }

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        EventBus.getInstance().register(this);
    }

    @Override
    protected void onPause() {
        super.onPause();
        EventBus.getInstance().unregister(this);
    }

    public void sendMessage(View view) {
        EventBus.getInstance().post("Hello EventBus");
    }
}

结果：