东莞招聘网站有哪些,建设电子商务网站需要什么设备,厦门seo推广,文学网站开发设计报告Google Guava EventBus(事件总线)的使用和源码的简单解析
什么是EventBus#xff1f;
事件总线#xff08;EventBus#xff09;是一种广泛用于软件架构中的设计模式#xff0c;用于实现解耦和松散耦合的通信机制。它可以帮助组织和管理应用程序中不同组件之间的通信…Google Guava EventBus(事件总线)的使用和源码的简单解析
什么是EventBus
事件总线EventBus是一种广泛用于软件架构中的设计模式用于实现解耦和松散耦合的通信机制。它可以帮助组织和管理应用程序中不同组件之间的通信以提高应用程序的可维护性、可扩展性和灵活性。
在事件总线模式中不同的组件通过订阅和发布事件来进行通信。发布者发布一个事件订阅者可以订阅该事件并在事件发生时执行相关的操作。事件总线充当了中介者的角色它负责管理所有的事件和订阅者并确保事件正确地传递到所有订阅者。
事件总线模式可以在很多场景中使用例如 Android 应用程序中的通信、分布式系统中的消息传递等。常见的事件总线实现包括 Google Guava 的 EventBus 和 Square 的 Otto。
总的来说EventBus就是应用了发布者/订阅者模式用来帮助我们进行各组件间通信的工具。我们这里解析的是Google Guava 的 EventBus它的官方文档在这里
EventBus三要素
github上的图: EventBus里有三个比较重要的概念:
Event 消息事件Publisher 消息发布者Subscriber 消息订阅者
乍一看这订阅者/发布者模式还和观察者模式有几分相似之处不同之处就在于EventBus的存在它作为一个中间件充当了消息传递的助手使得订阅者不必直接订阅发布者订阅事件总线即可。
EventBus的五种线程模型
EventBus的线程模型指的是根据事件的发布和订阅所在的线程决定应该在哪个线程中处理事件。EventBus中有以下四种线程模型 POSTING默认在发布事件的线程中执行即同步执行速度快。 MAIN在主线程UI线程中执行如果当前线程是主线程直接执行订阅方法否则通过主线程的Poster来执行。 BACKGROUND在后台线程中执行如果当前线程是主线程通过后台的Poster来执行否则直接执行订阅方法。 ASYNC在新的子线程中执行每个事件都会创建一个新的子线程即异步执行速度较慢。
除此之外还有一种模型 MAIN_ORDERED MAIN_ORDERED 模式也是运行在主线程上的模式不同于 MAIN 模式的是它保证了事件的顺序性。也就是说当一个事件在主线程中被发布时它会先进入一个队列中之后再一个个的被处理。这样可以保证相同事件类型的事件按照发送的顺序依次被执行。如果当前线程不是主线程那么它就直接被执行这也是为了避免在子线程中的事件被阻塞。
我们会在消息处理方法中利用Subscribe注解指定线程模型。
EventBus的简单使用
使用EventBus三步走
这里我们根据github上的三步分为四步走
定义Event事件类:
public class MessageEvent {private String message;public MessageEvent(String message){this.message message;}public String getMessage(){return message;}public void setMessage(String message){this.message message;}
}这里定义了一个MessageEvent用于传递事件。
声明订阅者注册订阅方法并且指定线程模型 Subscribe(threadMode ThreadMode.MAIN)//选择线程模型说明事件将在主线程中处理public void onMoonEvent(MessageEvent messageEvent){tv_message.setText(messageEvent.getMessage());}这里在MainActivity里注册了一个订阅方法消息处理方法指定了线程模型为MAIN说明订阅方法在主线程里执行。完整Demo在后面放出。
3.注册订阅者与订阅默认的事件总线 bt_subscription.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {//MainActivity注册了这条事件总线if(!EventBus.getDefault().isRegistered(MainActivity.this)){EventBus.getDefault().register(MainActivity.this);}}});这里先判断MainActivity是否已经与默认的事件总线订阅如果没有订阅就进行订阅。
4.发送消息事件–触发订阅方法 bt_message.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {EventBus.getDefault().post(new MessageEvent(欢迎来到德莱联盟));finish();}});这里我在第二个Activity里发送了一个消息事件到默认的事件总线中这样MainActivity中的订阅方法就会被触发。
2.小Demo
这里我贴出我的小Demo
MainActivity:
public class MainActivity extends AppCompatActivity {private TextView tv_message;private Button bt_message;private Button bt_subscription;Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_main);tv_message findViewById(R.id.tv_message);bt_message findViewById(R.id.bt_message);bt_subscription findViewById(R.id.bt_subscription);bt_message.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {startActivity(new Intent(MainActivity.this,SecondActivity.class));}});bt_subscription.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {//MainActivity注册了这条事件总线if(!EventBus.getDefault().isRegistered(MainActivity.this)){EventBus.getDefault().register(MainActivity.this);}}});}Overrideprotected void onDestroy() {super.onDestroy();EventBus.getDefault().unregister(this);}//通过参数类型来区分应该执行哪个方法Subscribe(threadMode ThreadMode.MAIN)//选择线程模型说明事件将在主线程中处理public void onMoonEvent(MessageEvent messageEvent){tv_message.setText(messageEvent.getMessage());}Subscribe(threadMode ThreadMode.MAIN,sticky true)public void ononStickyEvent(MessageEvent messageEvent){tv_message.setText(messageEvent.getMessage());}Subscribe(threadMode ThreadMode.MAIN)public void secondBack(String mes){Toast.makeText(this, mes, Toast.LENGTH_SHORT).show();}
}SecondActivity:
public class SecondActivity extends AppCompatActivity {TextView tv_message;Button bt_message;Button bt_sticky;Button mes;Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_second);tv_message findViewById(R.id.tv_message1);bt_message findViewById(R.id.bt_message1);bt_sticky findViewById(R.id.bt_sticky);mes findViewById(R.id.mes_event);bt_message.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {EventBus.getDefault().post(new MessageEvent(欢迎来到德莱联盟));finish();}});bt_sticky.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {EventBus.getDefault().postSticky(new MessageEvent(粘性事件));finish();}});mes.setOnClickListener(new View.OnClickListener() {Overridepublic void onClick(View v) {EventBus.getDefault().post(new String(另一个事件));finish();}});}
}MessageEvent:
//自定义的Event事件
public class MessageEvent {private String message;public MessageEvent(String message){this.message message;}public String getMessage(){return message;}public void setMessage(String message){this.message message;}
}https://github.com/MonsterTTL/Android-Impove/tree/master/EventBusDemo 具体的实例代码放在我的github上了,里面还涉及到了粘性事件以及多个订阅事件。
源码解析
getDefault()方法
先来看我们最常用的获取EventBus的方法 static volatile EventBus defaultInstance;public static EventBus getDefault() {EventBus instance defaultInstance;if (instance null) {synchronized (EventBus.class) {instance EventBus.defaultInstance;if (instance null) {instance EventBus.defaultInstance new EventBus();}}}return instance;}很显然这里用到了DCL单例模式确保缺省状态下EventBus的实例只有一个。我们顺着这个方法往下捋看new EventBus方法。
new EventBus()方法 private static final EventBusBuilder DEFAULT_BUILDER new EventBusBuilder();public EventBus() {this(DEFAULT_BUILDER);}EventBus(EventBusBuilder builder) {logger builder.getLogger();subscriptionsByEventType new HashMap();typesBySubscriber new HashMap();stickyEvents new ConcurrentHashMap();mainThreadSupport builder.getMainThreadSupport();mainThreadPoster mainThreadSupport ! null ? mainThreadSupport.createPoster(this) : null;backgroundPoster new BackgroundPoster(this);asyncPoster new AsyncPoster(this);indexCount builder.subscriberInfoIndexes ! null ? builder.subscriberInfoIndexes.size() : 0;subscriberMethodFinder new SubscriberMethodFinder(builder.subscriberInfoIndexes,builder.strictMethodVerification, builder.ignoreGeneratedIndex);logSubscriberExceptions builder.logSubscriberExceptions;logNoSubscriberMessages builder.logNoSubscriberMessages;sendSubscriberExceptionEvent builder.sendSubscriberExceptionEvent;sendNoSubscriberEvent builder.sendNoSubscriberEvent;throwSubscriberException builder.throwSubscriberException;eventInheritance builder.eventInheritance;executorService builder.executorService;}很显然获取default EventBus的调用流程是这样的:getDefault() - EventBus()-EventBus(DEFAULT_BUILDER)。说道Builder那么应该是采取建造者模式创建的我们接下来分析EventBusBuilder。
EventBusBuilder
builder里延伸下去的内容有点多我们直接先看EventBusBuilder的build方法
public EventBus build() {return new EventBus(this);}很显然该类的作用是构建一个EventBus实例并提供一些配置选项。该类具有多个属性如是否记录订阅者异常、是否发送没有订阅者事件等以及设置自定义线程池和日志处理程序等。类中的各种方法可用于配置这些属性例如logSubscriberExceptions()方法可用于设置是否记录订阅者异常。 这里我们就不关注打印日志记录异常等配置了我们关注具体的创建过程。
我们先来看看以下几个可选的方法
/*** 设置 EventBus 是否开启事件类继承的支持。默认情况下EventBus会考虑事件类继承关系即父类有订阅者时子类的订阅者也会被
通知。但是关闭这个特性可以提高事件发布的速度。如果事件类的继承层级比较复杂关闭继承支持的性能提升应该会大于20%。但需要注意
的是事件发布通常只占应用程序内部CPU时间的一小部分除非高速发布事件例如每秒钟发布数百/数千个事件。 ***/public EventBusBuilder eventInheritance(boolean eventInheritance) {this.eventInheritance eventInheritance;return this;}/*** 这是用来指定线程池类型的除了使用Builder内部自带的线程池我们也可以使用自己传入的线程池*/public EventBusBuilder executorService(ExecutorService executorService) {this.executorService executorService;return this;}/*** 这是用来跳过指定订阅者的方法验证的当你注册一个类作为事件订阅者时EventBus会自动检查该类是否具有合法的事件订阅方法。* 这些方法必须是公共的没有返回值只有一个参数并且在方法上需要使用Subscribe注解。然而有些情况下你可能希望让某* 个类作为订阅者但是它并不符合这些要求。例如它可能有一些不是用Subscribe注解的方法但是你仍然希望这些方法能够被* EventBus识别并调用。在这种情况下你可以使用skipMethodVerificationFor()方法来跳过验证允许这些方法被注册为事件订阅 * 方法。*/public EventBusBuilder skipMethodVerificationFor(Class? clazz) {if (skipMethodVerificationForClasses null) {skipMethodVerificationForClasses new ArrayList();}skipMethodVerificationForClasses.add(clazz);return this;}/** 允许在生成了索引的情况下也强制使用反射进行订阅者的查找。默认情况下如果有生成的索引则会使用它来查找订阅者而不是使用反射。通过调用这个方法并将其参数设置为 true可以强制使用反射进行查找。 */public EventBusBuilder ignoreGeneratedIndex(boolean ignoreGeneratedIndex) {this.ignoreGeneratedIndex ignoreGeneratedIndex;return this;}/** 开启严格方法验证 . 在 EventBus 中严格检查是指在注册和订阅事件时检查相关方法的参数和返回类型是否正确匹配以确保事件可以正确地被分发。如果开启了严格检查那么在注册或订阅事件时如果发现有方法的参数或返回类型与事件类型不匹配EventBus 会抛出一个异常防止程序出现意料之外的错误。*/public EventBusBuilder strictMethodVerification(boolean strictMethodVerification) {this.strictMethodVerification strictMethodVerification;return this;}/** 用于添加 SubscriberInfoIndex 对象到 EventBusBuilder 中。SubscriberInfoIndex 接口用于提供订阅者类和订阅方法信息的索引以便 EventBus 能够快速找到订阅者及其订阅方法。在 EventBus 中 */public EventBusBuilder addIndex(SubscriberInfoIndex index) {if (subscriberInfoIndexes null) {subscriberInfoIndexes new ArrayList();}subscriberInfoIndexes.add(index);return this;}
接下来看下比较重要的几个方法 public EventBus installDefaultEventBus() {synchronized (EventBus.class) {if (EventBus.defaultInstance ! null) {throw new EventBusException(Default instance already exists. It may be only set once before its used the first time to ensure consistent behavior.);}EventBus.defaultInstance build();return EventBus.defaultInstance;}}这个方法是用来安装默认EventBus里的defaultInstance的
class EventBus{
...static volatile EventBus defaultInstance;//默认的实例private static final EventBusBuilder DEFAULT_BUILDER new EventBusBuilder();//默认的Builder
...
}这段代码是通过构建者模式构建一个默认的EventBus实例并安装在默认的EventBus上只能在第一次使用默认的EventBus之前调用否则会抛出EventBusException异常。在synchronized块内首先判断是否已经存在默认的EventBus实例如果存在则抛出异常否则构建EventBus实例并将其设置为默认实例最后返回默认实例。
接下来看另一个方法 MainThreadSupport getMainThreadSupport() {if (mainThreadSupport ! null) {return mainThreadSupport;} else if (AndroidComponents.areAvailable()) {return AndroidComponents.get().defaultMainThreadSupport;} else {return null;}}这个方法是用来获取主线程支持的那什么是主线程支持呢我们点进去看看
public interface MainThreadSupport {boolean isMainThread();Poster createPoster(EventBus eventBus);
}
这里我们可以看到MainThreadSupport是一个接口定义了isMainThread和createPoster方法根据这两个方法名我们可以推测这个接口的作用就是用来支持消息在主线程里传递的isMainThread用于判断当前线程是否是主线程createPoster用于创建一个poster以在主线程里传递消息。
当然这只是一个接口我们要看具体的实现还是得折回去看getMainThreadSupport方法我们看这一段 else if (AndroidComponents.areAvailable()) {return AndroidComponents.get().defaultMainThreadSupport;}这段代码的作用就是判断当前平台是不是Android平台如果是Android平台就调用Android.get().defaultMainThreadSupport获取默认的主线程支持类。那这里又得涉及到AndroidComponents类了我们开一个子标题讨论这个类。
AndroidComponents类
AndroidComponents类是一个抽象类内容比较短这里先贴出它的全部源码
public abstract class AndroidComponents {private static final AndroidComponents implementation;static {implementation AndroidDependenciesDetector.isAndroidSDKAvailable()? AndroidDependenciesDetector.instantiateAndroidComponents(): null;}public static boolean areAvailable() {return implementation ! null;}public static AndroidComponents get() {return implementation;}public final Logger logger;public final MainThreadSupport defaultMainThreadSupport;public AndroidComponents(Logger logger, MainThreadSupport defaultMainThreadSupport) {this.logger logger;this.defaultMainThreadSupport defaultMainThreadSupport;}
}
这里将其声明为抽象的意义应该就是防止其实例化因为AndroidComponents的具体实现是由AndroidDependenciesDetector来决定的这个类会检测当前是否在Android环境下如果是则返回实际的实现否则返回null。而AndroidDependenciesDetector这个类是不允许被外部直接实例化的因此AndroidComponents也应该遵循同样的规则。
真正重要的代码部分是在静态代码块内代码块在初始化时会会检测当前是否在Android环境下如果是则返回实际的实现否则返回null。
接下来我们再深入向下扒找到MainThreadSupport的具体实现类再往下看我们可以发现这个类是由反射机制实现的不过我们先不关心这个直接找到具体实现类DefaultAndroidMainThreadSupport直接看它的源码
public class DefaultAndroidMainThreadSupport implements MainThreadSupport {Overridepublic boolean isMainThread() {return Looper.getMainLooper() Looper.myLooper();}Overridepublic Poster createPoster(EventBus eventBus) {return new HandlerPoster(eventBus, Looper.getMainLooper(), 10);}
}可以发现这个类实际上十分简单通过Looper判断是否是主线程接下来看他的Poster类。 具体的Poster类
public class HandlerPoster extends Handler implements Poster {private final PendingPostQueue queue;private final int maxMillisInsideHandleMessage;private final EventBus eventBus;private boolean handlerActive;public 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);}}}}
...我们先来看这前半部分在构造方法中新出现了一个queue是一个post队列是用于存储post请求的队列具体的PendingPostQueue类我们就不在具体看了只要知道其内部使用链表的结构连接的而PendingPostQueue类里记录了这个post队列的头部和尾部。
enqueue方法中将发送的Post请求进行入队如果handler不处于活跃状态就将其置位活跃状态然后获取一个Message对象并将其发送给MessageQueue中接下来我们看它是如何处理post请求的
public void handleMessage(Message msg) {boolean rescheduled false;try {long started SystemClock.uptimeMillis();while (true) {PendingPost pendingPost queue.poll();if (pendingPost null) {synchronized (this) {// Check again, this time in synchronizedpendingPost 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;}}由于HandlerPoster本身就是一个Handler所以上一步发送的Message实际上就是由它自己处理的就是在这个方法中处理Post请求具体是在这里执行的
eventBus.invokeSubscriber(pendingPost);这个方法会将Event事件推送到具体的订阅者类中去执行接下来我们重新回到EventBus类中看看这个过程是怎样执行的。
Subscription类
在正式介绍EventBus是怎样执行推送之前我们需要先了解两个前置类的信息首先我们先来看Subscriptiont类
final class Subscription {final Object subscriber;final SubscriberMethod subscriberMethod;/*** Becomes false as soon as {link EventBus#unregister(Object)} is called, which is checked by queued event delivery* {link EventBus#invokeSubscriber(PendingPost)} to prevent race conditions.*/volatile boolean active;Subscription(Object subscriber, SubscriberMethod subscriberMethod) {this.subscriber subscriber;this.subscriberMethod subscriberMethod;active true;}Overridepublic boolean equals(Object other) {if (other instanceof Subscription) {Subscription otherSubscription (Subscription) other;return subscriber otherSubscription.subscriber subscriberMethod.equals(otherSubscription.subscriberMethod);} else {return false;}}Overridepublic int hashCode() {return subscriber.hashCode() subscriberMethod.methodString.hashCode();}
}这个类里面的具体的SubscriberMethod和Object subscriber就不详细介绍了根据名字我们也可以知道final Object subscriber 存储的是订阅事件的对象而 final SubscriberMethod subscriberMethod 则包含了订阅事件的方法的信息例如该方法的名称、参数类型等等。通过这两个属性EventBus 可以将事件正确地分发给订阅者的对应方法。
PendingPost类
然后我们再来看PendingPost类
final class PendingPost {private final static ListPendingPost pendingPostPool new ArrayListPendingPost();Object event;Subscription subscription;PendingPost next;...}我们看到这个类里有一个静态的变量
private final static ListPendingPost pendingPostPool new ArrayListPendingPost();就是说所有的PendingPost类的实例都会共享这一个池子这个池子主要是用来优化GC性能的用于重用已创建的对象避免了大量的对象创建和销毁减小了内存开销。
1.构造方法 private PendingPost(Object event, Subscription subscription) {this.event event;this.subscription subscription;}其创建方法主要是传入发送的事件和订阅信息相关的Subscription然后将其包装成一个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;return pendingPost;}}return new PendingPost(event, subscription);}obtainPendingPost方法是根据pendingPostPool这个共享池来创建PendingPost对象如果共享池里还有可用对象就复用这个对象来创建新的PendingPost对象否则就直接创建一个新的对象。 static void releasePendingPost(PendingPost pendingPost) {pendingPost.event null;pendingPost.subscription null;pendingPost.next null;synchronized (pendingPostPool) {// Dont let the pool grow indefinitelyif (pendingPostPool.size() 10000) {pendingPostPool.add(pendingPost);}}}最后是release方法这个方法是用来将PendingPost对象释放会共享池中的不过这个共享池的大小也是有限制的最大不能超过10000。
EventBus类中如何推送事件
上面说到具体是通过invokeSubscriber这个方法来推送事件的事不宜迟我们马上来看看这个方法的内容 void invokeSubscriber(PendingPost pendingPost) {Object event pendingPost.event;Subscription subscription pendingPost.subscription;PendingPost.releasePendingPost(pendingPost);if (subscription.active) {invokeSubscriber(subscription, event);}}可以发现具体又调用了invokeSubscriber(subscription, event);方法我们紧接着看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);}}很显然这里是通过了反射机制subscription.subscriberMethod.method.invoke(subscription.subscriber, event)表示将event事件发送到subscriber中并触发subscriber的处理方法。method是SubscriberMethod中的一个字段表示订阅方法本身通过invoke方法来调用这个方法并将subscriber作为方法的调用者event作为参数传递进去。到这里我们就看完Event事件是如何传递给订阅者的了。
Register订阅事件总线
接下来我们看第二个内容订阅事件总线时做了什么
public void register(Object subscriber) {...Class? subscriberClass subscriber.getClass();ListSubscriberMethod subscriberMethods subscriberMethodFinder.findSubscriberMethods(subscriberClass);synchronized (this) {for (SubscriberMethod subscriberMethod : subscriberMethods) {subscribe(subscriber, subscriberMethod);}}}核心方法在于subscribe方法 private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {Class? eventType subscriberMethod.eventType;Subscription newSubscription new Subscription(subscriber, subscriberMethod);CopyOnWriteArrayListSubscription subscriptions subscriptionsByEventType.get(eventType);//获取同一个类型的消息事件下的所有Subscrption对象if (subscriptions null) {subscriptions new CopyOnWriteArrayList();subscriptionsByEventType.put(eventType, subscriptions);//如果原来还没有注册这一类型的消息事件则新创建一个List并加入} 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) {if (i size || subscriberMethod.priority subscriptions.get(i).subscriberMethod.priority) {subscriptions.add(i, newSubscription);break;}}//这一段是用来将新的subscription对象插入到同一类型的消息事件队列的合适位置中--将订阅者与注册方法关联起来ListClass? subscribedEvents typesBySubscriber.get(subscriber);if (subscribedEvents null) {subscribedEvents new ArrayList();typesBySubscriber.put(subscriber, subscribedEvents);}subscribedEvents.add(eventType);//这一段是用来将订阅者与消息事件类型关联起来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 - ListClass).SetMap.EntryClass?, Object entries stickyEvents.entrySet();for (Map.EntryClass?, 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);}}}具体来说subscribe方法做了三件事情 1. 维护了当前EventBus中的subscriptionsByEventType表 2. 维护了当前EventBus中的 typesBySubscriber表 3. 处理了粘性事件
详细的说明我放在上面的注释里面了我们先说维护的两个表第一个表是代表消息事件类型和Subscription关联的一个表 第二个表是表示订阅者Subscriber和消息事件类型关联的两个表这两个表可以在发布事件时可以快速地找到对应的订阅者并调用其处理方法。
然后是粘性事件的处理这里有一个标志位eventInheritance这个标志位我们在之前提到过我们可以回到EventBuilder的内容查看主要是涉及到继承问题的除去继承的问题粘性事件主要就是调用 checkPostStickyEventToSubscription(newSubscription, stickyEvent);直接点开这个方法会发现其调用了 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 dont take care of here.postToSubscription(newSubscription, stickyEvent, isMainThread());}}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 subscriberinvokeSubscriber(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);}}
实际上是调用了postToSubscription方法该方法根据回调方法指定的线程模型来传递粘性事件至于上面提到的三种poster我们之后再解析。
Post方法
订阅者注册完毕后我们就要用Post方法进行消息的发送了Post方法又可以分为post和postSticky接下来我们先看post方法 public void post(Object event) {PostingThreadState postingState currentPostingThreadState.get();ListObject eventQueue postingState.eventQueue;//获取发送线程的eventQueueeventQueue.add(event);//将当前要发送时间添加进eventQueue中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()) {//eventQueue不为空postSingleEvent(eventQueue.remove(0), postingState);//通过postSingleEvent方法依次将eventQueue中的事件发送}} finally {postingState.isPosting false;postingState.isMainThread false;}}}这里又冒出来了一个PostingThreadState我们看它是从哪里来的 private final ThreadLocalPostingThreadState currentPostingThreadState new ThreadLocalPostingThreadState() {Overrideprotected PostingThreadState initialValue() {return new PostingThreadState();}};
...final static class PostingThreadState {final ListObject eventQueue new ArrayList();boolean isPosting;boolean isMainThread;Subscription subscription;Object event;boolean canceled;}这里涉及到一个ThreadLocal类我们先来看看 在Java中ThreadLocal是一个非常有用的类它允许您在单个线程中存储和检索数据而不会与其他线程共享。ThreadLocal通过创建一个副本来解决线程安全问题每个线程都有自己的副本所以每个线程都可以独立地改变自己的副本而不会影响其他线程。 ThreadLocal提供了三个主要方法 get() - 返回当前线程的变量副本如果有如果没有则返回默认值。 set(T value) - 设置当前线程的变量副本。 remove() - 移除当前线程的变量副本。 ThreadLocal通常用于在单个线程中存储和检索上下文信息例如数据库连接、事务对象等。使用ThreadLocal可以避免在多线程环境下对这些资源进行同步操作从而提高性能。 所以在这里的语境下PostingThreadState就是每个发送post请求线程中的局部变量且各个发送线程之间是不会共享这个PostingThreadState变量的每个发送post请求的线程都会有自己私有的PostingThreadState变量。
post方法做的大致如下 先获取到当前发送线程的PostingThreadState变量 接着获取当前线程要发送的事件队列eventQueue然后将当前要发送的事件添加进入eventQueue队列中 如果当前发送线程处于空闲状态就将eventQueue中的事件通过postSingleEvent方法依次发送 既然说调用了postSingleEvent方法接下来我们理所当然地查看postSingleEvent方法 private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {Class? eventClass event.getClass();boolean subscriptionFound false;if (eventInheritance) {ListClass? 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));}}}这里又涉及到了eventInheritance标志位这里我们不管这个标志位查看这个方法的核心在于 subscriptionFound | postSingleEventForEventType(event, postingState, clazz);我们接着查看这个postSingleEventForEventType方法
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class? eventClass) {CopyOnWriteArrayListSubscription subscriptions;synchronized (this) {subscriptions subscriptionsByEventType.get(eventClass);}if (subscriptions ! null !subscriptions.isEmpty()) {for (Subscription subscription : subscriptions) {postingState.event event;postingState.subscription subscription;boolean aborted;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;}这个方法接收了发送的事件event发送线程的局部变量postingState还有事件类型eventClass。接下来它获取了对应事件类型下的subscriptions队列对其中的每个subscription对象调用了postToSubscription方法如果一切成功就返回true否则返回false。至于postToSubscription这个方法我们在之前也提到过就是 private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) 上次提到它是用来传递粘性事件当然它也可以用来传递普通事件这个方法将根据指定的线程模型来选择如何传递事件给订阅者回调处理。
至于粘性事件的POST 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 immediatelypost(event);}不同点就在于在执行post方法之前还将其加入到了粘性事件队列stickyEvents以便在其他订阅者完成注册时能接收到粘性事件。
三种Poster
之前提到过EventBus里有三种Poster在MainThreadSuppot中我们已经认识了HandlerPoster接下来我们看看其他两种Poster这里我们为了方便贴出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 {// temporary: technically not correct as poster not decoupled from subscriberinvokeSubscriber(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);}}很显然根据不同的线程模型该方法会采取的不同的策略和postermainThreadPoster就是我们一开始解析过的HandlerPoster接下来先看backgroundPoster private final PendingPostQueue queue;private final EventBus eventBus;private volatile boolean executorRunning;BackgroundPoster(EventBus eventBus) {this.eventBus eventBus;queue new PendingPostQueue();}BackgroundPoster内部会有自己的一条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);}}}这里BackgroundPoster用了同步代码块说明虽然Background模型是在子线程中运行的但是其会遵循顺序依次执行同一个发送队列发送的事件触发的多个Background线程模型的回调方法不能同时运行。
至于任务该如何执行eventBus中会有一个线程池这个线程池类型是根据builder决定的默认情况下这个线程池的类型将会是CachedThreadPool。接下来再看它的run方法
public void run() {try {try {while (true) {PendingPost pendingPost queue.poll(1000);if (pendingPost null) {synchronized (this) {// Check again, this time in synchronizedpendingPost 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;}}run方法很简单就是不断将queue中的PendingPost取出放入线程池中执行eventBus的invokeSubscriber方法 接下来我们看AsncPoster
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);}Overridepublic void run() {PendingPost pendingPost queue.poll();if(pendingPost null) {throw new IllegalStateException(No pending post available);}eventBus.invokeSubscriber(pendingPost);}AsyncPoster和BackgroundPoster其实有点像不同之处就在于AsyncPoster的post请求不必顺序执行多个回调可以同时进行。
总结
到此为止我们就已经解析了EventBus源码的核心基础部分了让我们总结一下流程图 这里的流程图是简化的版本这里具体实现的细节被隐藏起来了主要是帮助我们理解机制流程。
