文章目录

1 EventBus 简单使用
2 简单介绍

2.1 Subscribe
2.2 threadMode
2.3 register
2.4 post
2.5 unregister

3 源码分析

3.1 注册流程
3.2 发送流程

4 手写实现

EventBus可以代替Android传统的Intent,Handler,Broadcast或接口函数,在Fragment,Activity,Service线程之间传递数据，执行方法

github：https://github.com/greenrobot/EventBus

1 EventBus 简单使用

implementation 'org.greenrobot:eventbus:3.1.1'

import org.greenrobot.eventbus.EventBus;import org.greenrobot.eventbus.Subscribe;import org.greenrobot.eventbus.ThreadMode;public class MainActivity extends AppCompatActivity {    private TextView textView;        @Override    protected void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_main);        EventBus.getDefault().register(this);        textView = (TextView) findViewById(R.id.tv);    }    public void change(View view) {        startActivity(new Intent(this, SecondActivity.class));    }    @Subscribe(threadMode = ThreadMode.MAIN)    public void receive(Friend friend){        textView.setText(friend.toString() + "==" + Thread.currentThread().getName());        Toast.makeText(this, friend.toString() , Toast.LENGTH_SHORT).show();    }        @Override    protected void onDestroy() {        super.onDestroy();        EventBus.getDefault().unregister(this);    }}

public class SecondActivity extends AppCompatActivity {    @Override    protected void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_second);    }    public void send(View view) {        EventBus.getDefault().post(new Friend("hongxue", 18));    }    @Override    protected void onDestroy() {        super.onDestroy();    }}

接下来看一种情况：

将threadMode = ThreadMode.ASYNC

    @Subscribe(threadMode = ThreadMode.ASYNC)    public void receive(Friend friend){        textView.setText(friend.toString() + "==" + Thread.currentThread().getName());        Toast.makeText(this, friend.toString() , Toast.LENGTH_SHORT).show();    }

然后修改send方法，将post放入到一个线程中

 public void send(View view) {        new Thread(new Runnable() {            @Override            public void run() {                EventBus.getDefault().post(new Friend("hongxue", 18));            }        }).start();    }

点击发送，报错如下：

Could not dispatch event: class com.hongx.eventbus.Friend to subscribing classclass com.hongx.eventbus.MainActivityjava.lang.RuntimeException: Can't toast on a thread that has not called Looper.prepare()

这是为什么呢？

2 简单介绍

可以看到，发布者(Publisher)使用post()方法将Event发送到Event Bus，而后Event Bus自动将Event发送到多个订阅者(Subcriber)

2.1 Subscribe

@Subscribe采用运行时注解,且注解只能用在函数上,默认的threadmode为posting

2.2 threadMode


POSTING	默认的模式,开销最小的模式,因为声明为POSTING的订阅者会在发布的同一个线程调用，发布者在主线程那么订阅者也就在主线程,反之亦,避免了线程切换,如果不确定是否有耗时操作，谨慎使用，因为可能是在主线程发布
MAIN	主线程调用，视发布线程不同处理不同,如果发布者在主线程那么直接调用(非阻塞式),如果发布者不在主线程那么阻塞式调用
MAIN_ORDERED	和MAIN差不多,主线程调用，和MAIN不同的是他保证了post是非阻塞式的(默认走MAIN的非主线程的逻辑，所以可以做到非阻塞)
BACKGROUND	在子线程调用,如果发布在子线程那么直接在发布线程调用,如果发布在主线程那么将开启一个子线程来调用,这个子线程是阻塞式的,按顺序交付所有事件,所以也不适合做耗时任务,因为多个事件共用这一个后台线程
ASYNC	在子线程调用,总是开启一个新的线程来调用,适用于做耗时任务,比如数据库操作,网络请求等，不适合做计算任务,会导致开启大量线程

2.3 register

通过getDefault获取EventBus实例，这是一条默认的事件总线,通过单例模式实现,其构造函数是Public的也可以通过new的形式来创建多条事件总线
从缓存中获取订阅方法列表,如果缓存中不存在则通过反射获取到订阅者所有的函数,遍历再通过权限修饰符.参数长度(只允许一个参数).注解(@Subscribe)来判断是否是具备成为订阅函数的前提,具备则构建一个SubscriberMethod(订阅方法，其相当于一个数据实体类,包含方法,threadmode,参数类型,优先级,是否粘性事件这些参数),循环结束订阅函数列表构建完成添加进入缓存
对subscriptionsByEventType.typesBySubscriber完成数据初始化,subscriptionsByEventType根据参数类型存储订阅者和订阅方法,typesBySubscriber根据订阅者存储了所有的参数类型，subscriptionsByEventType主要是post时使用,因为其存储了订阅者和订阅事件这两个参数在反射时要用到,typesBySubscriber在反注册时可以根据订阅者获取到存储的事件类型就可以从subscriptionsByEventType中获取到对应的订阅者和订阅方法释放资源,还可以用来判断是否注册,这些最后会分析到

2.4 post

使用ThreadLocal来存储事件,他可以隔离多个线程对数据的访问冲突

根据事件类型(也就是Post参数的类型)为key从subscriptionsByEventType获取订阅方法和订阅者,这个容器不了解的可以看下注册的总结

根据线程和threadmode来判断

2.5 unregister

根据订阅者从typesBySubscriber中获取到所有的事件类型
有了事件类型列表就可以遍历事件类型从subscriptionsByEventType获取到对应的订阅者包括订阅函数来释放
根据订阅者删除掉typesBySubscriber中的事件类型此时订阅者已经被释放不再持有该订阅者引用

3 源码分析

3.1 注册流程

1 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);            }        }    }

2 查找订阅者方法

subscriberMethods ：为存储所有订阅函数的集合

先从缓存中读取并返回subscriberMethods
如果缓存中没有读取到，就去查找，将subscriberMethods 保存到缓存中并返回

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;        }    }

通过debug，我们可以看到如下输出结果：

3.2 发送流程

post方法源码如下：

  public void post(Object event) {        PostingThreadState postingState = currentPostingThreadState.get();        List<Object> eventQueue = postingState.eventQueue;        eventQueue.add(event);        if (!postingState.isPosting) {            postingState.isMainThread = isMainThread();            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;            }        }    }

currentPostingThreadState 是一个ThreadLocal，ThreadLocal可以解决多线程的并发访问问题，他会为每一个线程提供一个独立的变量副本，可以隔离多个对数据访问的冲突。

private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {        @Override        protected PostingThreadState initialValue() {            return new PostingThreadState();        }    };

获取事件类型：

  //获取事件类型  private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {        Class<?> eventClass = event.getClass();        boolean subscriptionFound = false;        if (eventInheritance) {            List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);            int countTypes = eventTypes.size();            for (int h = 0; h < countTypes; h++) {                Class<?> clazz = eventTypes.get(h);                subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);            }        } else {            subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);        }        if (!subscriptionFound) {            if (logNoSubscriberMessages) {                logger.log(Level.FINE, "No subscribers registered for event " + eventClass);            }            if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&                    eventClass != SubscriberExceptionEvent.class) {                post(new NoSubscriberEvent(this, event));            }        }    }

这里根据注册时候总结这个这个容器中装的是订阅者和订阅方法，现在根据发送事件的类型来获取到对应的订阅者和订阅方法这些参数是反射必须要用到的。

    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;    }

debug查看结果如下图所示：

    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 MAIN_ORDERED:                if (mainThreadPoster != null) {                    mainThreadPoster.enqueue(subscription, event);                } else {                    // temporary: technically not correct as poster not decoupled from subscriber                    invokeSubscriber(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);        }    }

最后Subscription中包含了订阅者和订阅方法，event是post的参数，这里通过反射直接调用订阅者的订阅方，完成本次通信。

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);        }    }

4 手写实现

import android.os.Handler;import android.os.Looper;import java.lang.reflect.InvocationTargetException;import java.lang.reflect.Method;import java.util.ArrayList;import java.util.HashMap;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.Set;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class HxEventbus {    private static HxEventbus instance = new HxEventbus();    private Map<Object, List<SubscribleMethod>> cacheMap;    private Handler handler;    //线程池    private ExecutorService executorService;    public static HxEventbus getDefault() {        return instance;    }    private HxEventbus() {        this.cacheMap = new HashMap<>();        handler = new Handler(Looper.getMainLooper());        executorService = Executors.newCachedThreadPool();    }    //注册    public void register(Object subscriber) {        Class<?> aClass = subscriber.getClass();        List<SubscribleMethod> subscribleMethods = cacheMap.get(subscriber);        //如果已经注册，就不需要注册        if (subscribleMethods == null) {            subscribleMethods = getSubscribleMethods(subscriber);            cacheMap.put(subscriber, subscribleMethods);        }    }    //遍历能够接收事件的方法    private List<SubscribleMethod> getSubscribleMethods(Object subscriber) {        List<SubscribleMethod> list = new ArrayList<>();        Class<?> aClass = subscriber.getClass();        //需要 subscriber --- 》BaseActivity    ------->Activitiy        while (aClass != null) {            //判断分类是在那个报下，（如果是系统的就不需要）            String name = aClass.getName();            if (name.startsWith("java.") ||                    name.startsWith("javax.") ||                    name.startsWith("android.") ||                    name.startsWith("androidx.")) {                break;            }            Method[] declaredMethods = aClass.getDeclaredMethods();            for (Method method : declaredMethods) {                HxSubscribe annotation = method.getAnnotation(HxSubscribe.class);                if (annotation == null) {                    continue;                }                //检测这个方法合不合格                Class<?>[] parameterTypes = method.getParameterTypes();                if (parameterTypes.length != 1) {                    throw new RuntimeException("eventbusz只能接收一个参数");                }                //符合要求                HxThreadMode dnThreadMode = annotation.threadMode();                SubscribleMethod subscribleMethod = new SubscribleMethod(method, dnThreadMode, parameterTypes[0]);                list.add(subscribleMethod);            }            aClass = aClass.getSuperclass();        }        return list;    }    //取消注册    public void unregister(Object subscriber) {        Class<?> aClass = subscriber.getClass();        List<SubscribleMethod> list = cacheMap.get(subscriber);        //如果获取到        if (list != null) {            cacheMap.remove(subscriber);        }    }    public void post(final Object obj) {        Set<Object> set = cacheMap.keySet();        Iterator<Object> iterator = set.iterator();        while (iterator.hasNext()) {            //拿到注册类            final Object next = iterator.next();            //获取类中所有添加注解的方法            List<SubscribleMethod> list = cacheMap.get(next);            for (final SubscribleMethod subscribleMethod : list) {                //判断这个方法是否应该接收事件                if (subscribleMethod.getEventType().isAssignableFrom(obj.getClass())) {                    switch (subscribleMethod.getThreadMode()) {                        case MAIN:                            //如果接收方法在主线程执行的情况                            if(Looper.myLooper() == Looper.getMainLooper()){                                invoke(subscribleMethod, next, obj);                            } else {                                //post方法执行在子线程中，接收消息在主线程中                                handler.post(new Runnable() {                                    @Override                                    public void run() {                                        invoke(subscribleMethod, next, obj);                                    }                                });                            }                            break;                        //接收方法在子线程种情况                        case ASYNC:                            //post方法执行在主线程中                            if(Looper.myLooper() == Looper.getMainLooper()){                                executorService.execute(new Runnable() {                                    @Override                                    public void run() {                                        invoke(subscribleMethod, next, obj);                                    }                                });                            } else {                                //post方法执行在子线程中                                invoke(subscribleMethod, next, obj);                            }                            break;                        case POSTING:                            break;                    }                }            }        }    }    private void invoke(SubscribleMethod subscribleMethod, Object next, Object obj) {        Method method = subscribleMethod.getMethod();        try {            method.invoke(next, obj);        } catch (IllegalAccessException e) {            e.printStackTrace();        } catch (InvocationTargetException e) {            e.printStackTrace();        }    }}

public enum HxThreadMode {    POSTING,    MAIN,    MAIN_ORDERED,    BACKGROUND,    ASYNC}

import java.lang.annotation.ElementType;import java.lang.annotation.Retention;import java.lang.annotation.RetentionPolicy;import java.lang.annotation.Target;@Target(ElementType.METHOD)@Retention(RetentionPolicy.RUNTIME)public @interface HxSubscribe {    HxThreadMode threadMode() default HxThreadMode.POSTING;}

import java.lang.reflect.Method;//注册类中的注册方法信息public class SubscribleMethod {    //注册方法    private Method method;    //线程类型    private HxThreadMode threadMode;    //参数类型    private Class<?> eventType;    public SubscribleMethod(Method method, HxThreadMode threadMode, Class<?> eventType) {        this.method = method;        this.threadMode = threadMode;        this.eventType = eventType;    }    public Method getMethod() {        return method;    }    public void setMethod(Method method) {        this.method = method;    }    public HxThreadMode getThreadMode() {        return threadMode;    }    public void setThreadMode(HxThreadMode threadMode) {        this.threadMode = threadMode;    }    public Class<?> getEventType() {        return eventType;    }    public void setEventType(Class<?> eventType) {        this.eventType = eventType;    }}

import android.content.Intent;import android.support.v7.app.AppCompatActivity;import android.os.Bundle;import android.view.View;import android.widget.TextView;import android.widget.Toast;import com.hongx.eventbus.core.HxEventbus;import com.hongx.eventbus.core.HxSubscribe;import com.hongx.eventbus.core.HxThreadMode;public class MainActivity extends AppCompatActivity {    private TextView textView;    @Override    protected void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_main);        HxEventbus.getDefault().register(this);        textView = (TextView) findViewById(R.id.tv);    }        public void change(View view) {        startActivity(new Intent(this, SecondActivity.class));    }    @HxSubscribe(threadMode = HxThreadMode.MAIN)    public void receive(Friend friend){        textView.setText(friend.toString() + "==" + Thread.currentThread().getName());        Toast.makeText(this, friend.toString() , Toast.LENGTH_SHORT).show();    }    @Override    protected void onDestroy() {        super.onDestroy();        HxEventbus.getDefault().unregister(this);    }}

import android.os.Bundle;import android.support.v7.app.AppCompatActivity;import android.view.View;import com.hongx.eventbus.core.HxEventbus;import org.greenrobot.eventbus.EventBus;public class SecondActivity extends AppCompatActivity {    @Override    protected void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_second);    }    public void send(View view) {        HxEventbus.getDefault().post(new Friend("hongxue", 18));    }    @Override    protected void onDestroy() {        super.onDestroy();    }}

Android(安卓)EventBus 架构设计