EventBus

关于Eventbus的问题

1.线程只要非UI线程和非UI线程就可以了,为什么EventBus中要有好几种Threadmode呢?这有什么好处?

2.EventBus的post方法是怎么调用相应register的相应方法的?

4月18日重新又看下代码

1private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
2        switch (subscription.subscriberMethod.threadMode) {
3            case PostThread:
4                invokeSubscriber(subscription, event);
5                break;
6            case MainThread:
7                if (isMainThread) {
8                    invokeSubscriber(subscription, event);
9                } else {
10                    mainThreadPoster.enqueue(subscription, event);
11                }
12                break;
13            case BackgroundThread:
14                if (isMainThread) {
15                    backgroundPoster.enqueue(subscription, event);
16                } else {
17                    invokeSubscriber(subscription, event);
18                }
19                break;
20            case Async:
21                asyncPoster.enqueue(subscription, event);
22                break;
23            default:
24                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
25        }
26    }

可见PostThread模式的意思是post事件的时候在哪类线程,最终就在哪类线程调用方法. mainThread模式当post的时候不在主线程,是通过mainThreadPost.enqueue去执行的. 看mainThreadPost为啥能执行.

final class HandlerPoster extends Handler

private final HandlerPoster mainThreadPoster;

EventBus(EventBusBuilder builder) { //省略 mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10); //省略 } 可以发现这里使用了Looper.getMainLooer(); 就是说让HandlerPoster这个Handler的handleMessage方法会在主线程执行. 这是enqueue方法的内容

1     void enqueue(Subscription subscription, Object event) {
2        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
3        synchronized (this) {
4            queue.enqueue(pendingPost);
5            if (!handlerActive) {
6                handlerActive = true;
7                if (!sendMessage(obtainMessage())) {        //A处
8                    throw new EventBusException("Could not send handler message");
9                }
10            }
11        }
12    }

可以看到A处代码sendMessage. 然后就到了handleMessage方法.

1 @Override
2    public void handleMessage(Message msg) {
3        boolean rescheduled = false;
4        try {
5            long started = SystemClock.uptimeMillis();
6            while (true) {
7                PendingPost pendingPost = queue.poll();
8                if (pendingPost == null) {
9                    synchronized (this) {
10                        // Check again, this time in synchronized
11                        pendingPost = queue.poll();
12                        if (pendingPost == null) {
13                            handlerActive = false;
14                            return;
15                        }
16                    }
17                }
18                eventBus.invokeSubscriber(pendingPost); //B处
19                long timeInMethod = SystemClock.uptimeMillis() - started;
20                if (timeInMethod >= maxMillisInsideHandleMessage) {
21                    if (!sendMessage(obtainMessage())) {
22                        throw new EventBusException("Could not send handler message");
23                    }
24                    rescheduled = true;
25                    return;
26                }
27            }
28        } finally {
29            handlerActive = rescheduled;
30        }
31    }    

所以B处代码一定是在主线程执行的了.

而Async模式

1 case Async:
2        asyncPoster.enqueue(subscription, event);
3        break;
4
5
6    class AsyncPoster implements Runnable {
7        public void enqueue(Subscription subscription, Object event) {
8        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
9        queue.enqueue(pendingPost);
10        eventBus.getExecutorService().execute(this);
11    }
12    }

就是把这个任务放到一个线程池执行.所以必须Async了.

1   case BackgroundThread:
2        if (isMainThread) {
3            backgroundPoster.enqueue(subscription, event);
4        } else {
5            invokeSubscriber(subscription, event);
6        }

看一下这个BackGroundThread如果post事件的时候在主线程.利用backGroundPoster.enqueue去执行. 也就是把这个事件放到一个线程池去执行. 如果判断出post的时候不是主线程.则直接执行.那么当前就不在主线程上必然就是background了.

这里的线程池默认是这样 private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();

4月18日重新又看下代码

`

1private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
2        switch (subscription.subscriberMethod.threadMode) {
3            case PostThread:
4                invokeSubscriber(subscription, event);
5                break;
6            case MainThread:
7                if (isMainThread) {
8                    invokeSubscriber(subscription, event);
9                } else {
10                    mainThreadPoster.enqueue(subscription, event);
11                }
12                break;
13            case BackgroundThread:
14                if (isMainThread) {
15                    backgroundPoster.enqueue(subscription, event);
16                } else {
17                    invokeSubscriber(subscription, event);
18                }
19                break;
20            case Async:
21                asyncPoster.enqueue(subscription, event);
22                break;
23            default:
24                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
25        }
26    }

`

private static final String ON_EVENT_METHOD_NAME = "onEvent";

注册的时候最终会调用的方法

1 private synchronized void register(Object subscriber, boolean sticky, int priority) {
2        //A strart
3        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());
4        //A end
5        for (SubscriberMethod subscriberMethod : subscriberMethods) {
6            subscribe(subscriber, subscriberMethod, sticky, priority);
7        }
8 }

1.首先通过反射拿到带包名的类名,跳过java,javax,android开头的类. 2.然后通过反射拿到方法的修饰符等.过滤掉非public方法.判断方法以指定字符串开头,并且方法的参数只能有一个等判断 3.然后通过截取字符串知道方法的threadMode. 4.然后攒出SubscriberMethod对象. 5.等for循环结束后这里就能拿到一个Lis<SubscriberMethod> subscriberMethods列表. 6.另外有一个缓存 private static final Map<String, List<SubscriberMethod>> methodCache = new HashMap<String, List<SubscriberMethod>>(); 这里有一个缓存,key是订阅者的全类名,value为全部找到的SubscriberMethod的列表.

在A处的代码如下:

1 List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
2        String key = subscriberClass.getName();
3        List<SubscriberMethod> subscriberMethods;
4        synchronized (methodCache) {
5            subscriberMethods = methodCache.get(key);
6        }
7        if (subscriberMethods != null) {
8            return subscriberMethods;
9        }
10
11        subscriberMethods = new ArrayList<SubscriberMethod>();
12        Class<?> clazz = subscriberClass;
13        HashSet<String> eventTypesFound = new HashSet<String>();
14        StringBuilder methodKeyBuilder = new StringBuilder();
15        while (clazz != null) {
16            String name = clazz.getName();
17            if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("android.")) {
18                // Skip system classes, this just degrades performance
19                break;
20            }
21
22            // Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)
23        Method[] methods = clazz.getDeclaredMethods();
24        for (Method method : methods) 
25        {
26            String methodName = method.getName();
27            if (methodName.startsWith(ON_EVENT_METHOD_NAME)) 
28            {
29                int modifiers = method.getModifiers();
30                //在java.lang.reflect包中有一个Modifier.java这么一个类,这个类中定义了一些常量表示方法的修饰符.
31                //这里的意思就是public方法.
32                if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) 
33                {
34                    //这里通过反射拿到方法的参数所对应的Class
35                    Class<?>[] parameterTypes = method.getParameterTypes();
36                    //然后限制方法的参数只能有一个
37                    if (parameterTypes.length == 1) 
38                    {
39                        //根据方法名截取字符串知道方法的threadMode.
40                        String modifierString = methodName.substring(ON_EVENT_METHOD_NAME.length());
41                        ThreadMode threadMode;
42                        //默认threadMode为postThread即post的是什么线程就是什么线程
43                        if (modifierString.length() == 0) 
44                        {
45                            threadMode = ThreadMode.PostThread;
46                        } else if (modifierString.equals("MainThread")) 
47                        {
48                            threadMode = ThreadMode.MainThread;
49                        } else if (modifierString.equals("BackgroundThread")) 
50                        {
51                            threadMode = ThreadMode.BackgroundThread;
52                        } else if (modifierString.equals("Async")) 
53                        {
54                            threadMode = ThreadMode.Async;
55                        }
56                        else
57                        {
58                            if (skipMethodVerificationForClasses.containsKey(clazz)) {
59                                continue;
60                            } else {
61                                throw new EventBusException("Illegal onEvent method, check for typos: " + method);
62                            }
63                        }
64
65                        //这里拿到方法参数对应的Class对象
66                        Class<?> eventType = parameterTypes[0];
67                        methodKeyBuilder.setLength(0);
68                        methodKeyBuilder.append(methodName);
69                        methodKeyBuilder.append('>').append(eventType.getName());
70                        //这里的methodKey为方法名>方法参数全类名
71                        String methodKey = methodKeyBuilder.toString();
72
73                        //hashSet的add方法有返回值,如果加入成功为true.
74                        //这个的eventTypesFound为一个HashSet<String>
75                        if (eventTypesFound.add(methodKey)) 
76                        {
77                            // Only add if not already found in a sub class
78                            //这个的method为方法对应的反射Method对象.
79                            subscriberMethods.add(new SubscriberMethod(method, threadMode, eventType));
80                        }
81                    }
82                } else if (!skipMethodVerificationForClasses.containsKey(clazz)) {
83                    Log.d(EventBus.TAG, "Skipping method (not public, static or abstract): " + clazz + "."
84                            + methodName);
85                }
86            }
87        }
88
89        //当for循环结束就拿到一个List<SubscriberMethod>对象
90            clazz = clazz.getSuperclass();
91
92        }
93
94        if (subscriberMethods.isEmpty()) 
95        {
96            throw new EventBusException("Subscriber " + subscriberClass + " has no public methods called "
97                    + ON_EVENT_METHOD_NAME);
98        }
99        else 
100        {
101            synchronized (methodCache) {
102                methodCache.put(key, subscriberMethods);
103            }
104
105            return subscriberMethods;
106        }
107    }

然后就到了EventBus类中的subscribe方法 1.subscriptionsByEventType的声明如下: private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType; 即一个维护了方法参数Class对象和一个Subscription列表的映射 Subscription对象是一个维护订阅者,订阅者方法对象,和订阅优先级的对象. 因为第一次会put进subscriptionsByEventType的映射,所以以后再注册就是重复注册了.

2. typesBySubscriber的声明如下: private final Map<Object, List<Class<?>>> typesBySubscriber; 维护一个订阅者和订阅方法参数Class对象列表的映射. 3.sticky的情况先略过.

   // Must be called in synchronized block private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) { //这里是方法的参数的Class对象 Class<?> eventType = subscriberMethod.eventType;

   1    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
   2
   3    //Subsciption是一个维护订阅者,订阅的方法对象和订阅优先级的对象.
   4    Subscription newSubscription = new Subscription(subscriber, subscriberMethod, priority);
   5
   6    if (subscriptions == null) {
   7        subscriptions = new CopyOnWriteArrayList<Subscription>();
   8        subscriptionsByEventType.put(eventType, subscriptions);
   9    }
   10    else 
   11    {
   12        if (subscriptions.contains(newSubscription)) {
   13            throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
   14                    + eventType);
   15        }
   16    }
   17
   18    // Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)
   19    // subscriberMethod.method.setAccessible(true);
   20
   21    //如果新的订阅者的优先级更高,那么放到subscriptions列表的更前面一位
   22    int size = subscriptions.size();
   23    for (int i = 0; i <= size; i++) {
   24        if (i == size || newSubscription.priority > subscriptions.get(i).priority) {
   25            subscriptions.add(i, newSubscription);
   26            break;
   27        }
   28    }
   29
   30    //private final Map<Object, List<Class<?>>> typesBySubscriber;
   31    //维护一个订阅者和订阅方法参数对象列表的映射.
   32
   33    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
   34    if (subscribedEvents == null) {
   35        subscribedEvents = new ArrayList<Class<?>>();
   36        typesBySubscriber.put(subscriber, subscribedEvents);
   37    }
   38    subscribedEvents.add(eventType);
   39
   40    if (sticky) {
   41        Object stickyEvent;
   42        synchronized (stickyEvents) {
   43            stickyEvent = stickyEvents.get(eventType);
   44        }
   45        if (stickyEvent != null) {
   46            // If the subscriber is trying to abort the event, it will fail (event is not tracked in posting state)
   47            // --> Strange corner case, which we don't take care of here.
   48            postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
   49        }
   50    }
   51}
   

然后看post方法怎么把post的东西在相应订阅者身上调用.

1 /** Posts the given event to the event bus. */
2public void post(Object event) 
3{
4    PostingThreadState postingState = currentPostingThreadState.get();
5    List<Object> eventQueue = postingState.eventQueue;
6    //给当前线程的PostingThreadState对象赋值,这里的值是要发送的事件.
7    eventQueue.add(event);
8
9    if (!postingState.isPosting) 
10    {
11        postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
12        postingState.isPosting = true;
13        if (postingState.canceled) {
14            throw new EventBusException("Internal error. Abort state was not reset");
15        }
16        try 
17        {
18            while (!eventQueue.isEmpty()) 
19            {
20                postSingleEvent(eventQueue.remove(0), postingState);
21            }
22        }
23        finally {
24            postingState.isPosting = false;
25            postingState.isMainThread = false;
26        }
27    }
28}

1.这里的currentPostingThreadState对象的声明如下: private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() { @Override protected PostingThreadState initialValue() { return new PostingThreadState(); } };

就是一个ThreadLocal里面存了PostingThreadState,确保获取每一个线程自己的PostingThreadState. 这个PostingThreadState的声明如下:

1 /** For ThreadLocal, much faster to set (and get multiple values). */
2    final static class PostingThreadState {
3        final List<Object> eventQueue = new ArrayList<Object>();
4        boolean isPosting;
5        boolean isMainThread;
6        Subscription subscription;
7        Object event;
8        boolean canceled;
9    }

然后就到了postSingEvent方法.

1 private void postSingleEvent(Object event, PostingThreadState postingState) throws Error 
2 {
3        Class<?> eventClass = event.getClass();
4        boolean subscriptionFound = false;
5        if (eventInheritance) 
6        {
7            List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
8            int countTypes = eventTypes.size();
9            for (int h = 0; h < countTypes; h++) {
10                Class<?> clazz = eventTypes.get(h);
11                subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
12            }
13        }
14        else
15        {
16            subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
17        }
18
19        if (!subscriptionFound) {
20            if (logNoSubscriberMessages) {
21                Log.d(TAG, "No subscribers registered for event " + eventClass);
22            }
23            if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
24                    eventClass != SubscriberExceptionEvent.class) {
25                post(new NoSubscriberEvent(this, event));
26            }
27        }
28 }

有一个eventInheritance变量标识事件是不是继承的.然后根据是否做不同的处理. 先看下不是继承的情况.

1 private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) 
2 {
3        CopyOnWriteArrayList<Subscription> subscriptions;
4        synchronized (this) 
5        {
6         
7            //这里的eventClass是EventBus所post的事件的class对象
8
9            //即一个维护了方法参数Class对象和一个Subscription列表的映射
10            //Subscription对象是一个维护订阅者,订阅者方法对象,和订阅优先级的对象.
11            subscriptions = subscriptionsByEventType.get(eventClass);
12        }
13
14        //到这里就能拿到订阅这个事件的所有subscription对象.
15
16        if (subscriptions != null && !subscriptions.isEmpty()) 
17        {
18            for (Subscription subscription : subscriptions) 
19            {
20                postingState.event = event;
21                postingState.subscription = subscription;
22                boolean aborted = false;
23                try 
24                {
25                    //这个时候postingState已经有了订阅者,订阅的方法,是否在主线程,要发送的事件对象等信息.
26                    postToSubscription(subscription, event, postingState.isMainThread);
27                    aborted = postingState.canceled;
28                }
29                finally 
30                {
31                    postingState.event = null;
32                    postingState.subscription = null;
33                    postingState.canceled = false;
34                }
35                if (aborted) {
36                    break;
37                }
38            }
39
40            return true;
41        }
42
43        return false;
44    }

然后就到了postToSubscription方法 根据不同的threadMode区分调用.

1private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) 
2{
3        switch (subscription.subscriberMethod.threadMode)
4         {
5            case PostThread:
6                invokeSubscriber(subscription, event);
7                break;
8            case MainThread:
9                if (isMainThread) 
10                {
11                    invokeSubscriber(subscription, event);
12                }
13                else 
14                {
15                    mainThreadPoster.enqueue(subscription, event);
16                }
17                break;
18
19            case BackgroundThread:
20                if (isMainThread) 
21                {
22                    backgroundPoster.enqueue(subscription, event);
23                }
24                else 
25                {
26                    invokeSubscriber(subscription, event);
27                }
28                break;
29
30            case Async:
31                asyncPoster.enqueue(subscription, event);
32                break;
33            default:
34                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
35        }
36    }

如果是postThread或者是MainThread的话调用invokdeSubscriber方法. 这里很简单,就是一个反射调用.该有的信息都有了.

1void invokeSubscriber(Subscription subscription, Object event) {
2        try {
3            subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
4        } catch (InvocationTargetException e) {
5            handleSubscriberException(subscription, event, e.getCause());
6        } catch (IllegalAccessException e) {
7            throw new IllegalStateException("Unexpected exception", e);
8        }
9    }

如果是BackgroundThread并且是是是是在主线程中调用的话. backgroundPoster.enqueue(subscription, event); 这里边做了什么?

enqueue中最终会调用eventBus.getExecutorService().execute(this); 大概是有一个线程池去执行这个BackgroundPoster.执行的具体任务看run方法 最终有一个eventBus.invokeSubscriber(pendingPost);

void invokeSubscriber(PendingPost pendingPost) { Object event = pendingPost.event; Subscription subscription = pendingPost.subscription; PendingPost.releasePendingPost(pendingPost); if (subscription.active) { invokeSubscriber(subscription, event); } }

然后是反射调用: 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); } }

1 final class BackgroundPoster implements Runnable {
2
3    private final PendingPostQueue queue;
4    private final EventBus eventBus;
5
6    private volatile boolean executorRunning;
7
8    BackgroundPoster(EventBus eventBus) {
9        this.eventBus = eventBus;
10        queue = new PendingPostQueue();
11    }
12
13    public void enqueue(Subscription subscription, Object event) {
14        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
15        synchronized (this) {
16            queue.enqueue(pendingPost);
17            if (!executorRunning) {
18                executorRunning = true;
19                eventBus.getExecutorService().execute(this);
20            }
21        }
22    }
23
24    @Override
25    public void run() {
26        try {
27            try {
28                while (true) {
29                    PendingPost pendingPost = queue.poll(1000);
30                    if (pendingPost == null) {
31                        synchronized (this) {
32                            // Check again, this time in synchronized
33                            pendingPost = queue.poll();
34                            if (pendingPost == null) {
35                                executorRunning = false;
36                                return;
37                            }
38                        }
39                    }
40                    eventBus.invokeSubscriber(pendingPost);
41                }
42            } catch (InterruptedException e) {
43                Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);
44            }
45        } finally {
46            executorRunning = false;
47        }
48    }
49}