【死磕【Sharding-jdbc】---EventBus-轻量级进程内事件分发组件

@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    }}

/** * @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);    }}

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