在上一篇文章中可能你的EventBus使用并不正确,是时候真正搞懂EventBus了(上),我们说了使用apt方式进行优化使用eventbus,并对EventBus#regi线程池sterEventBus#unRegister方法进行了说明。对这块内线程池的使用容不熟悉的,可以先看下在继续看下面的内容。

本节内容将讲解剩下的内容,包括了EventBus#post粘性事件等。

1、EventBus#post(Object event)

之前说了,register的时候,会将查找到的添加的注解的方法进行查找,并保存。这节将讲解使用。

public void post(Object event) {
   //获取PostingThreadState对象
   PostingThreadState postingState = currentPostingThreadState.get();
   //获取eventQueue,并将event设置进去
   List<Object> eventQueue = postingState.eventQueue;
   eventQueue.add(event);
   //判断当前的event是否被post
   if (!postingState.isPosting) {
       //设置当前post是否主线程
       postingState.isMainThread = isMainThread();
       //将任务标志为已经post
       postingState.isPosting = true;
       if (postingState.canceled) {
           throw new EventBusException("Internal error. Abort state was not reset");
       }
       try {
           while (!eventQueue.isEmpty()) {
               //这里的eventQueue是一个List,
               // 而我们知道List#remove方法是从队列中移除对象,并返回这个移除的对象
               postSingleEvent(eventQueue.remove(0), postingState);
           }
       } finally {
           postingState.isPosting = false;
           postingState.isMainThread = false;
       }
   }
}

其实这里面核心代码主要就是postSingleEvent(even线程池tQueue.remove(0), postingState)

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    Class<?> eventClass = event.getClass();
    boolean subscriptionFound = false;
    //eventInheritance 默认为true,表示会检索继承关系
    if (eventInheritance) {
        //根据event类对象查找所有的eventType类型(包含了所有的event的父类对象)
        List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
        int countTypes = eventTypes.size();
        for (int h = 0; h < countTypes; h++) {
            Class<?> clazz = eventTypes.get(h);
            //判断是否检索到eventType, 并发送post出去
            subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
        }
    } else {
        subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
    }
    if (!subscriptionFound) {
        //如果没有找到的话 && logNoSubscriberMessages == true, 则抛出异常,否则发送NoSubscriberEvent
        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));
        }
    }
}

代码逻辑很简单,使用postSingleEventForEventType发送事件,并且发挥发送结果,如果发送结果返回false,则判断logNoSubscriberMessages 是否为true,true的话跑出没有找到注册方法的异常

接着看p线程数越多越好吗ostSing优先级队列leEventForEventType

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
     CopyOnWriteArrayList<Subscription> subscriptions;
     synchronized (this) {
         //还记得之前在EventBus.getDefault().register()执行到最后的subscribe方法吗,
         // 将注册方法的event(也就是我们设置@Subscribe注解的方法参数)作为key,
         // 将有优先级顺序的方法结合作为value,这边直接遍历执行,
         //subscription里面之前就已经包含了method对象,直接invoke执行,就达到了调用消息传递的目标了
         subscriptions = subscriptionsByEventType.get(eventClass);
     }
     if (subscriptions != null && !subscriptions.isEmpty()) {
         for (Subscription subscription : subscriptions) {
             postingState.event = event;
             postingState.subscription = subscription;
             boolean aborted;
             try {
                 //这里是重点!!,还记得之前@Subscribe注解中定义了threadMode吗,这里就派上用场了
                 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;
 }

可以看到使用到线程池原理了之前register的时候的subscr线程和进程的区别是什么iptionsByEventType,如果看过上优先级英文篇文章,就知道它是一个以event类方法为key,所有添优先级排序加了注解参数方法中参数为even线程撕裂者t的方法集合为value的map集合。

然后进行遍历,调用postToSubscription(subscription, event, postingState.isMainThread);触达事件线程池拒绝策略发送。

再看postToSubscription(subscription, event, p线程数是什么ostingState.isMainThread);方法实现。

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    //根据注解方法上的threadMode来区分使用那种poster发送event出去
    switch (subscription.subscriberMethod.threadMode) {
        case POSTING:
            invokeSubscriber(subscription, event);
            break;
        case MAIN:
            if (isMainThread) {
                //invokeSubscriber本身不关心线程,在当前线程直接发送
                invokeSubscriber(subscription, event);
            } else {
                //使用handler切换到主线程执行
                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);
    }
}

这里其实核心用到了三个post线程池拒绝策略er和5种Threa线程池核心参数dMode

首先看三种poster

  • mainThreadPoster

这个poster本身就比较简单了线程池参数配置,它的createPoster(EventBus eventBu线程池核心参数s) 方法实际上就是HandlerPoster,并且在创建HandlerPoster的时候,使用的是MainLooper,是不是秒懂了,其实就是不管线程池的七个参数你是在哪个线程发消息,最后都是从主线程出去的。

  • backgroun线程是什么意思dPoster

这个poster是继承自Runnable,在你每次send的时候,将event会添加到一个队列中,然优先级英文后丢到线程池中调用执行。简单列一下核心代码。

final class BackgroundPoster implements Runnable, Poster {
    private final PendingPostQueue queue;
    private final EventBus eventBus;
    private volatile boolean executorRunning;
    BackgroundPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }
    //调用的时候,会塞到队列中
    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!executorRunning) {
                executorRunning = true;
                eventBus.getExecutorService().execute(this);
            }
        }
    }
    @Override
    public void run() {
        try {
            try {
                while (true) {
                    PendingPost pendingPost = queue.poll(1000);
                    if (pendingPost == null) {
                        synchronized (this) {
                            // Check again, this time in synchronized
                            pendingPost = queue.poll();
                            if (pendingPost == null) {
                                executorRunning = false;
                                return;
                            }
                        }
                    }
                    eventBus.invokeSubscriber(pendingPost);
                }
            } catch (InterruptedException e) {
                eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
            }
        } finally {
            executorRunning = false;
        }
    }
}
  • asyncPoster

这个poster和上面的b线程池的使用ackgrou线程池面试题dPoster,核心思路是一样的线程和进程的区别是什么,也都是通过线程池,只是区别是backgro优先级最高的运算符undPoster每次发送一条,会做线程同步和标志位判断,确保执行顺序。

class AsyncPoster implements Runnable, Poster {
    private final PendingPostQueue queue;
    private final EventBus eventBus;
    AsyncPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }
    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        queue.enqueue(pendingPost);
        eventBus.getExecutorService().execute(this);
    }
    @Override
    public void run() {
        PendingPost pendingPost = queue.poll();
        if(pendingPost == null) {
            throw new IllegalStateException("No pending post available");
        }
        eventBus.invokeSubscriber(pendingPost);
    }
}

其实除了上面的三种poster,你可能还看到有这么一个方法invokeSubscriber(Subscription su线程池bscription, Object 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);
    }
}

其实上面的三种不管是同步、异步线程池面试题线程池的七个参数还是主线程,最终线程和进程的区别是什么执行都会是这个方法。

线程池创建搞明白了三种poster和最终的通知方式之后,对于5中ThreadMode就不难理解了。

5种T优先级越小越优先吗hreadMode分别是

  • POSTING 在哪个线程发送,直接线程池参数配置哪个线程接收
  • MAIN 不管是从哪个线程发送,最终都回到主线程使用
  • MAIN_ORDERED 主线程使用Handler顺序发送
  • BACKGROUND 不管是主线程还是子线程发送,最终都回到子线程使用
  • ASY线程撕裂者NC 根本不判断线程,直接丢给子线程发送给,最终回到子线程使用

至此,相信你对于EventBus中的从注册、反注册、发送的逻辑使用应该都很清除了吧。下面还有一个(postSticky(Obje线程池面试题ct event))没有说到,当然线程池的工作原理工作中好像不是太常用,但是理解线程池的七个参数学习下并没有啥坏处是不是。

2、粘性事件pos线程数越多越好吗tSticky(Object e线程池有哪几种vent)

 public void postSticky(Object event) {
     synchronized (stickyEvents) {
         stickyEvents.put(event.getClass(), event);
     }
     // Should be posted after it is putted, in case the subscriber wants to remove immediately
     post(event);
 }

本身逻辑是不是很简单,向stickyEvents中加进去了event,然线程数是什么后调用了我们刚讲过的post方法。至于stickyEvents很容易就能看出来,就是一个map。至于用户,再线程池的工作原理往下看。

大家还记得上篇文章中,说到postSingleEventForEventType方法的时候,有一线程池的使用部分是粘性事件,我说后面再讲吗。下面就讲那段代码拎出来。

EventBus#subscribe

...
//subscribe方法
if (subscriberMethod.sticky) {
    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);
    }
}
...

这里用到线程池了一个checkPostStickyEventToSubscription方法

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, isMainThread());
    }
}

最终线程池参数配置执行了postToSubscription(newSubscription, sticky优先级Event, isMainThread());方法,是不是很熟悉,刚才刚讲过的内容。

相信你对postSticky使用原理已经想明白了,我最后就归纳总结下吧

post的时候,会对event做一个保存,再次声明周期过来的时候,会在register的时候进行查找,如果是粘性事件,则从stickyEvent 取出来,直接执行post流程。