EventBus来自于google-guava包中。源码注释如下:
Dispatches events to listeners, and provides ways for listeners to register themselves. The EventBus allows publish-subscribe-style communication between components without requiring the components to explicitly register with one another (and thus be aware of each other). It is designed exclusively to replace traditional Java in-process event distribution using explicit registration. It is not a general-purpose publish-subscribe system, nor is it intended for interprocess communication.
翻译:将事件分派给监听器,并为监听器提供注册自己的方法。EventBus允许组件之间的发布 - 订阅式通信,而不需要组件彼此明确注册(并且因此彼此意识到)。 它专门用于使用显式注册替换传统的Java进程内事件分发。 它不是一个通用的发布 - 订阅系统,也不是用于进程间通信。
关于EventBus的用例代码提取自sharding-jdbc源码,并结合lombok最大限度的简化:
@NoArgsConstructor(access = AccessLevel.PRIVATE)public final class EventBusInstance { private static final EventBus INSTANCE = new EventBus(); public static EventBus getInstance() { return INSTANCE; }}
@Getter@Setterpublic class DMLExecutionEvent { private String id; private String dataSource; private Date sendTime;}
public final class DMLExecutionEventListener { @Subscribe @AllowConcurrentEvents public void listener(final DMLExecutionEvent event) { System.out.println("监听的DML执行事件: " + JSON.toJSONString(event)); // do something }}
-- Main--主方法:注册订阅者监听事件,以及发布事件。
/** * @author wangzhenfei9 * @version 1.0.0 * @since 2018年04月24日 */public class Main { static{ System.out.println("register listener..."); EventBusInstance.getInstance().register(new DMLExecutionEventListener()); } public static void main(String[] args) throws InterruptedException { for (int i=0; i<10; i++) { pub(); Thread.sleep(3000); } } private static void pub(){ DMLExecutionEvent event = new DMLExecutionEvent(); event.setId(String.valueOf(new Random().nextInt(1000))); event.setDataSource("sj_db_1"); event.setSendTime(new Date()); System.out.println("发布的DML执行事件: " + JSON.toJSONString(event)); EventBusInstance.getInstance().post(event); }}
EventBus一些重要方法解释如下:
主要分析发布事件以及订阅的核心源码;
public void post(Object event) { // 得到所有该类已经它的所有父类(因为有些注册监听器是监听其父类) Set<Class<?>> dispatchTypes = flattenHierarchy(event.getClass()); boolean dispatched = false; // 遍历类本身以及所有父类 for (Class<?> eventType : dispatchTypes) { // 重入读锁先锁住 subscribersByTypeLock.readLock().lock(); try { // 得到类的所有订阅者,例如DMLExecutionEvent的订阅者就是DMLExecutionEventListener(EventSubscriber有两个属性:重要的属性target和method,target就是监听器即DMLExecutionEventListener,method就是监听器方法即listener;从而知道DMLExecutionEvent这个事件由哪个类的哪个方法监听处理) Set<EventSubscriber> wrappers = subscribersByType.get(eventType); if (!wrappers.isEmpty()) { // 如果有时间订阅者,那么dispatched = true,表示该事件可以分发 dispatched = true; // 遍历所有的时间订阅者,每个订阅者的队列都增加该事件 for (EventSubscriber wrapper : wrappers) { enqueueEvent(event, wrapper); } } } finally { subscribersByTypeLock.readLock().unlock(); } } if (!dispatched && !(event instanceof DeadEvent)) { post(new DeadEvent(this, event)); } // 分发进入队列的事件 dispatchQueuedEvents();}/** * queues of events for the current thread to dispatch; * 核心数据结构为LinkedList,保存的是EventBus.EventWithSubscriber类型数据 */private final ThreadLocal<Queue<EventBus.EventWithSubscriber>> eventsToDispatch = new ThreadLocal<Queue<EventBus.EventWithSubscriber>>() { @Override protected Queue<EventBus.EventWithSubscriber> initialValue() { return new LinkedList<EventBus.EventWithSubscriber>(); } };void enqueueEvent(Object event, EventSubscriber subscriber) { // 数据结构为new LinkedList<EventWithSubscriber>(),EventWithSubscriber就是对event和subscriber的封装,LinkedList数据结构保证进入队列和消费队列顺序一致 eventsToDispatch.get().offer(new EventBus.EventWithSubscriber(event, subscriber));}/** * Drain the queue of events to be dispatched. As the queue is being drained, * new events may be posted to the end of the queue. * 排干要被分发的事件队列,正在排干的过程中,可能有新的事件被追加到队列尾部 */void dispatchQueuedEvents() { // don't dispatch if we're already dispatching, that would allow reentrancy // and out-of-order events. Instead, leave the events to be dispatched // after the in-progress dispatch is complete. // 如果正在排干队列,则不分发 if (isDispatching.get()) { return; } // ThreadLocal设置正在分发即isDispatching为true isDispatching.set(true); try { Queue<EventBus.EventWithSubscriber> events = eventsToDispatch.get(); EventBus.EventWithSubscriber eventWithSubscriber; while ((eventWithSubscriber = events.poll()) != null) { // 调用订阅者处理事件(method.invoke(target, new Object[] { event });,method和target来自订阅者) dispatch(eventWithSubscriber.event, eventWithSubscriber.subscriber); } } finally { // ThreadLocal可能内存泄漏,用完需要remove isDispatching.remove(); // 队列中的事件任务处理完,清空队列,即所谓的排干(Drain) eventsToDispatch.remove(); }}
/** * Registers all subscriber methods on {@code object} to receive events. * 注册object上所有订阅方法,用来接收事件,上面的使用参考,DMLExecutionEventListener就是这里的object */public void register(Object object) { // Multimap是guava自定义数据结构,类似Map<K, Collection<V>>,key就是事件类型,例如DMLExecutionEvent,value就是EventSubscriber即事件订阅者集合(说明,这个的订阅者集合是指object里符合订阅者的所有方法,例如DMLExecutionEventListener.listener(),DMLExecutionEventListener中可以有多个订阅者,注解@Subscribe即可), Multimap<Class<?>, EventSubscriber> methodsInListener = finder.findAllSubscribers(object); // 重入写锁保证线程安全 subscribersByTypeLock.writeLock().lock(); try { // 把订阅者信息放到map中缓存起来(发布事件post()时就会用到) subscribersByType.putAll(methodsInListener); } finally { subscribersByTypeLock.writeLock().unlock(); }}
END