preface
EventBus is pretty simple to use, but what’s the logic behind it? Here is a simple start, starting from register to learn from internal snooping. The version of EventBus in this paper is 3.2.0.
1. Register to register
public void register(Object subscriber) {
// Get the subscriber Class and pass in the findSubscriberMethods method,
// The findSubscriberMethods method name tells you that this method is used to find which subscribers are approved
// Subscribe the Subscribe method of the annotation tag.Class<? > subscriberClass = subscriber.getClass(); List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);// The subscription methods found here are used for subscription-related processing.
synchronized (this) {
for(SubscriberMethod subscriberMethod : subscriberMethods) { subscribe(subscriber, subscriberMethod); }}}Copy the codeThe main logic of this method is to find subscription methods and process found subscription methods. Check out findSubscriberMethods() to see how it works.
1.1 findSubscriberMethods
List<SubscriberMethod> findSubscriberMethods(Class
subscriberClass) {
// First use METHOD_CACHE to look for cached subscription method information and return it if it does.
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if(subscriberMethods ! =null) {
return subscriberMethods;
}
// Two search methods are provided. If you do not customize, ignoreGeneratedIndex defaults to false
if (ignoreGeneratedIndex) {
// Find by reflection
subscriberMethods = findUsingReflection(subscriberClass);
} else {
// Search through the EventBus index
subscriberMethods = findUsingInfo(subscriberClass);
}
// It is not possible to register only for subscribers without providing a qualified @subscribe method
// Then the exception is thrown directly at runtime to abort program execution.
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {
// Find the target subscription method, put it in the cache to speed up next use and then return.
METHOD_CACHE.put(subscriberClass, subscriberMethods);
returnsubscriberMethods; }}Copy the codeEventBus provides two search methods: reflection and index. In particular, index search generates all subscriber information into a class at compile time, and can be retrieved directly from the index without reflection. The ignoreGeneratedIndex flag is false by default, meaning index lookup is used, but we need to configure it a little extra to actually use it.
android {
defaultConfig {
javaCompileOptions {
annotationProcessorOptions {
arguments = [ eventBusIndex : 'com.example.myapp.MyEventBusIndex' ]
}
}
}
}
dependencies {
def eventbus_version = '3.2.0'
implementation "org.greenrobot:eventbus:$eventbus_version"
annotationProcessor "org.greenrobot:eventbus-annotation-processor:$eventbus_version"
}
Copy the codeYou can see that the annotation handler is used to generate the index, specify the index generation class name, and then initialize the EventBus.
public class App extends Application {
@Override
public void onCreate(a) {
EventBus bus = EventBus.builder()
.addIndex(new MyEventBusIndex())
.build()
}
}
Copy the codeThen we can use the indexing function by replacing the default EventBus with our own EventBus.
EventBus provides two ways to do this, but let’s start with findUsingReflection().
1.2 findUsingReflection
private List<SubscriberMethod> findUsingReflection(Class
subscriberClass) {
// FindState is a lookup helper class, which is obtained by the prepareFindState method,
// It can be cached by EventBus and then called to subscriberClass to call FindState
// Initialize for this subscriber.
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
FindState. Clazz refers to the subscription class being searched. Use traversal because
// The subscription class may not be limited to the current subscriber, or it may be the parent class of the subscriber.
while(findState.clazz ! =null) {
// Reflection is used to find the method.
findUsingReflectionInSingleClass(findState);
// Switch to the next class to be searched
findState.moveToSuperclass();
}
// When all subscriber related subscription classes are searched, the subscriber method information will be returned.
return getMethodsAndRelease(findState);
}
Copy the codeLet’s start with the first paragraph, where we’re talking about the FindState class, so let’s go in prepareFindState().
private FindState prepareFindState(a) {
// private static final int POOL_SIZE = 4;
// private static final FindState[] FIND_STATE_POOL = new FindState[POOL_SIZE];
synchronized (FIND_STATE_POOL) {
for (int i = 0; i < POOL_SIZE; i++) {
FindState state = FIND_STATE_POOL[i];
if(state ! =null) {
FIND_STATE_POOL[i] = null;
returnstate; }}}return new FindState();
}
Copy the codeEventBus can cache up to four findStates. After all, this is a frequently used class. If you do not cache it, you will create a new object and return it.
Then focus on circulation within the search, first take a look at findState. MoveToSuperClass () this method, from the method name that it is the current cycle of subscription class switch for the parent class.
void moveToSuperclass(a) {
/ / skipSuperClasses default is false, in findState initForSubscriber (in) the assignment
// When it is true it breaks out of the external loop and ends the search for the subscriber.
if (skipSuperClasses) {
clazz = null;
} else {
// Get the parent class of the current subscriber
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; }}}Copy the codeYou can see that it is indeed traversal looking for the subscriber itself and its superclass, with the additional processing done here, namely skipping the system class.
Back to the external methods, key, we will go in findUsingReflectionInSingleClass () to understand the real search logic.
1.2.1 findUsingReflectionInSingleClass
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// Get all declared methods of the current subscription class
methods = findState.clazz.getDeclaredMethods();
} catch (Throwable th) {
// If an exception occurs, use getMethods() and skip the parent class lookup, i.e. the outer loop will loop only once.
methods = findState.clazz.getMethods();
findState.skipSuperClasses = true;
}
// Check each method to see if it is a valid subscription method.
for (Method method : methods) {
int modifiers = method.getModifiers();
// check if the subscription method is public, non-absctract, non-static and has only one parameter
// Subscribe is annotated with @subscribe.
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];
if (findState.checkAdd(method, eventType)) {
When method can be placed in findState, set threadMode and priority respectively
// and method are packaged into SubscribeMethod and put into findState. The // SubscribeMethod is very familiar and we can see that when we first enter the Register,
// It contains the subscription method itself and its accompanying threadMode and priority.
ThreadMode threadMode = subscribeAnnotation.threadMode();
findState.subscriberMethods.add(newSubscriberMethod(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"); }}}Copy the codeGetDeclaredMethods () ¶ getDeclaredMethods() ¶ getDeclaredMethods() ¶ getDeclaredMethods() ¶ getDeclaredMethods() ¶ But includes all methods that inherit implementations from superclasses and interfaces), and then validates the reflected methods. Go back to findUsingReflection(), and when all the methods are done iterating through, getMethodAndRelease() will be called, so check it out.
private List<SubscriberMethod> getMethodsAndRelease(FindState findState) {
List<SubscriberMethod> subscriberMethods = new ArrayList<>(findState.subscriberMethods);
findState.recycle();
synchronized (FIND_STATE_POOL) {
for (int i = 0; i < POOL_SIZE; i++) {
if (FIND_STATE_POOL[i] == null) {
FIND_STATE_POOL[i] = findState;
break; }}}return subscriberMethods;
}
Copy the codeAs you can see, you create a list to load all the SubscribeMethods you just found, then release FindState and try to add FInsState to the cache.
The reflection lookup ends here, and we’ll look at the lookup using indexes.
1.3 findUsingInfo
private List<SubscriberMethod> findUsingInfo(Class
subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while(findState.clazz ! =null) {
findState.subscriberInfo = getSubscriberInfo(findState);
if(findState.subscriberInfo ! =null) {
SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
for (SubscriberMethod subscriberMethod : array) {
if(findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) { findState.subscriberMethods.add(subscriberMethod); }}}else {
findUsingReflectionInSingleClass(findState);
}
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
private SubscriberInfo getSubscriberInfo(FindState findState) {
// This distinguishes the indexes between inherited relationships, while SubscriberInfo has superSubscriberInfo
// Super SubscriberInfo is returned only when the currently searched subscription class is the same as superSubscriberClass.
if(findState.subscriberInfo ! =null&& findState.subscriberInfo.getSuperSubscriberInfo() ! =null) {
SubscriberInfo superclassInfo = findState.subscriberInfo.getSuperSubscriberInfo();
if (findState.clazz == superclassInfo.getSubscriberClass()) {
returnsuperclassInfo; }}// The check above did not get the appropriate SubscriberInfo, so the subscriberInfoIndexes are iterated here
// Performs a lookup and returns when a matching SubscriberInfo is found for the currently searched subscription class.
if(subscriberInfoIndexes ! =null) {
for (SubscriberInfoIndex index : subscriberInfoIndexes) {
SubscriberInfo info = index.getSubscriberInfo(findState.clazz);
if(info ! =null) {
returninfo; }}}// The index cannot be found and null is returned, which means that reflection is needed outside.
return null;
}
Copy the codeAs can be seen, the index lookup method is not much different from the reflection method in code logic, and when this method cannot be found, reflection will be used to search again. Here look at the getSubscriberInfo() method, which obtains SubscriberInfo for the subscription class searched. SubscriberInfo contains information about the subscription class that is generated for us by the annotation processor at the compiler.
That’s the end of index lookup, and now you can go back to the main line, register().
1.4 the subscribe
The process of searching for subscriberMethods has been combed out previously, so now that these subscriberMethods are found, they need to be subscribed.
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
// This part of the code is mainly packaged for subscriber and subscriberMethod and generated
// Uniform Subscription, which is then cached using a list, is checked to see if it already exists,
// Throw exception if it already exists, i.e., do not allow multiple registrations of the same Subscription class
// overrides equals, internally using subscriber and scubscriberMethod for comparison.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); }}// The priority is set first.
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; }}// Caches subscription classes and related events to determine whether they have been registered and unregister later.List<Class<? >> subscribedEvents = typesBySubscriber.get(subscriber);if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
/ /... Dealing with sticky events will come later.
}
Copy the codeThis part of the subscription processing code is relatively simple and clear, the register analysis has come to an end, the following is a brief summary.
Conclusion: The registration process searches for relevant subscription information through subscribers. If there is cache, it will be directly used; otherwise, it will be searched through reflection or index. By default, the index will be used. Subscribers and events are cached to determine registration status and later unregistration.
2. Unregister
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 {
logger.log(Level.WARNING, "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) {
// Cancel the subscription activation status, if there are still events will not be able to receive.
subscription.active = false; subscriptions.remove(i); i--; size--; }}}}Copy the codeSearch in typesBySubscriber by subscriber, unregister related events after finding them, and remove related caches.
3. Post Sends common events
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; }}}Copy the codeCurrentPostingThreadState is a ThreadLocal, used to get the current thread private PostingThreadState, this class encapsulates the event queue, subscription information, and distributed control, After the event is enqueued, PostingThreadState determines whether the event is currently being distributed. If not, it starts the distribution, and sends individual events by traversing the event queue and calling postSingleEvent().
3.1 postSingleEvent
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)); }}}Copy the codeThis method mainly deals with two cases. If the event is allowed to have inheritance relationship (default is true, can be closed), all the parent classes and interfaces of the event are queried, and then the event related classes are traversed to send the event. If inheritance is not allowed, the event is sent directly. The return value of the method indicates whether the subscriber was found. If not, the default Log is printed and the NoSubscriberEvent is sent. Then look at postSingleEventForEventType ().
3.2 postSingleEventForEventType
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;
}
Copy the codeIf it is empty, it returns false. If it is not empty, it will iterate over the subscription information and start sending real events. Aborted here indicates the interruption of event distribution. Move on to postToSubscription.
3.3 postToSubscription
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 {
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); }}Copy the codeHere you see the handling for thread distribution. The default is POSTING when threadMode is not explicitly specified, that is, POSTING in the current thread.
-
POSTING
Call invokeSubscriber() to check it out.
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); }}private void handleSubscriberException(Subscription subscription, Object event, Throwable cause) { if (event instanceof SubscriberExceptionEvent) { if (logSubscriberExceptions) { logger.log(Level.SEVERE, "SubscriberExceptionEvent subscriber " + subscription.subscriber.getClass() + " threw an exception", cause); SubscriberExceptionEvent exEvent = (SubscriberExceptionEvent) event; logger.log(Level.SEVERE, "Initial event " + exEvent.causingEvent + " caused exception in "+ exEvent.causingSubscriber, exEvent.throwable); }}else { if (throwSubscriberException) { throw new EventBusException("Invoking subscriber failed", cause); } if (logSubscriberExceptions) { logger.log(Level.SEVERE, "Could not dispatch event: " + event.getClass() + " to subscribing class " + subscription.subscriber.getClass(), cause); } if (sendSubscriberExceptionEvent) { SubscriberExceptionEvent exEvent = new SubscriberExceptionEvent(this, cause, event, subscription.subscriber); post(exEvent); }}}Copy the codeCan be seen directly call subscribe method through reflection, then handle exceptions, handleSubscriberException () emits an SubscriberExceptionEvent by default.
-
MAIN
public class HandlerPoster extends Handler implements Poster { private final PendingPostQueue queue; private final int maxMillisInsideHandleMessage; private final EventBus eventBus; private boolean handlerActive; protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) { super(looper); this.eventBus = eventBus; this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage; queue = new PendingPostQueue(); } public void enqueue(Subscription subscription, Object event) { PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event); synchronized (this) { queue.enqueue(pendingPost); if(! handlerActive) { handlerActive =true; if(! sendMessage(obtainMessage())) {throw new EventBusException("Could not send handler message"); }}}}@Override public void handleMessage(Message msg) { boolean rescheduled = false; try { long started = SystemClock.uptimeMillis(); while (true) { PendingPost pendingPost = queue.poll(); if (pendingPost == null) { synchronized (this) { pendingPost = queue.poll(); if (pendingPost == null) { handlerActive = false; return; } } } eventBus.invokeSubscriber(pendingPost); long timeInMethod = SystemClock.uptimeMillis() - started; if (timeInMethod >= maxMillisInsideHandleMessage) { if(! sendMessage(obtainMessage())) {throw new EventBusException("Could not send handler message"); } rescheduled = true; return; }}}finally{ handlerActive = rescheduled; }}}// PendingPost static PendingPost obtainPendingPost(Subscription subscription, Object event) { synchronized (pendingPostPool) { int size = pendingPostPool.size(); if (size > 0) { PendingPost pendingPost = pendingPostPool.remove(size - 1); pendingPost.event = event; pendingPost.subscription = subscription; pendingPost.next = null; returnpendingPost; }}return new PendingPost(event, subscription); } Copy the codePendingPost is used in handleposter.enqueue () to load the event information to be sent. PendingPost is then enqueued and an empty Message is sent if the Handler is idle to activate Handler processing.
In handleMessage () will PendingPost out team and emit events, then judge maxMillisInsideHandleMessage time can also perform an incident to send, or sent over, MaxMillisInsideHandleMessage default 10 milliseconds.
As you can see, obtainPendingPost() also uses a cache pool to save memory, an idea that has seen a lot of use in EventBus.
In addition, the MAIN mode is not queued by default if the dispatch thread is on the MAIN thread, which means that the subscribed method needs to return quickly or else it will block the dispatch of the event, which further affects the caller of the external POST ().
-
MAIN_ORDER
PostToSubscription () is blocked by default (in the case of the MAIN thread), which means that the next event is not sent until the last event has been processed, whereas MAIN_ORDER defaults to enqueuing events to mainPoster, This mode ensures that the distribution thread is not blocked.
-
BACKGROUND
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(a) { 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; }}}// PendingPostQueue synchronized PendingPost poll(a) { PendingPost pendingPost = head; if(head ! =null) { head = head.next; if (head == null) { tail = null; }}return pendingPost; } synchronized PendingPost poll(int maxMillisToWait) throws InterruptedException { if (head == null) { wait(maxMillisToWait); } return poll(); } Copy the codeIf the current thread is the main thread is the team, or directly to send, after the team handled by thread pool, the thread pool is Executors newCachedThreadPool (). As you can see from the comments on the BACKGROUND thread mode, events are only ordered in a single thread, but how does this work internally with a thread pool that can have almost unlimited threads?
Looking directly at the run() method, you can see that the synchronized code block is used in the while loop and, crucially, queue.poll(1000), you can see that wait() is actually called to sleep the current thread. This is to prevent the thread from executing without a PendingPost (in which case there is no guarantee that a series of events will be handled in one thread), and then the thread wakes up and obtains execution rights and decides if null is fetched, Poll (), if there is no PendingPost there are no more events to process and the thread is returned. Otherwise, reflection calls the subscription method.
. See also understand why use this Executors newCachedThreadPool () it can also ensure that event distribution in a single thread, the reason is in the loop body using the dormancy mechanism, such as first-class team of events, and then traverse.
It is important to note that this mode does not necessarily cut the thread every time. If the thread is not distributed on the main thread, the subscription method will be called directly. Therefore, this mode will block, and the subscription method should be returned as quickly as possible.
-
ASYNC
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(a) { PendingPost pendingPost = queue.poll(); if(pendingPost == null) { throw new IllegalStateException("No pending post available"); } eventBus.invokeSubscriber(pendingPost); }}Copy the codeThis mode is similar to BACKGROUND mode, except that the pool is always enqueued, so that each time the pool is processed in a different thread. It ensures that each time the pool is stagged with the distributor thread. This mode does not block the distributor thread.
At this point, the normal event sending process ends.
4. PostSticky sends sticky events
public void postSticky(Object event) {
synchronized (stickyEvents) {
stickyEvents.put(event.getClass(), event);
}
post(event);
}
Copy the codeBasically, you store sticky events, and then you send a normal event. We already know that sticky events are handled in the registration process, so let’s go back.
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
/ /... Omit irrelevant code
if (subscriberMethod.sticky) {
if(eventInheritance) { 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); }}}private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
if(stickyEvent ! =null) { postToSubscription(newSubscription, stickyEvent, isMainThread()); }}Copy the codeIf the event is inheritable, it needs to go through all sticky events to find derived events to send, otherwise it can directly send. Then checkPostStickyEventToSubscription () is only a judgment event is not null and then enter the postToSubscription (), which has been analyzed in front of us.
This is the end of the sticky event flow.
Ending