手写事件发布/订阅框架(二)

一、背景

之前写了一篇《手写事件发布订阅框架》，虽然可以用但代码写的比较粗糙，且存在优化的空间，于是对其进行了重构主要包括以下几点：

面向接口编程，包结构更加清晰。
框架改成spring-boot-starter的形式实现即插即用。
对核心类EventManager的代码进行了部分剥离，使其更符合职责单一原则。
发布事件时事件类不用再继承Event类，用户可以随意自定义。
新的项目结构如下图：

二、代码介绍

之前的EventManager（现EventListenerManager）承载了很多不该属于它的功能，比如初始化监听器配置信息，操作Spring上下文等，优化后通过OnceApplicationContextEventListener来进行监听器配置的初始化，EventListenerManager只用于管理事件监听器。

EventListenerManager

package cn.sp.manager;

import cn.sp.domain.EventListenerRegistration;
import cn.sp.event.Event;
import cn.sp.listener.EventListener;

/**
 * @author Ship
 * @version 1.0.0
 * @description:
 * @date 2022/05/05 
 */
public interface EventListenerManager {


    /**
     * 注册一个事件监听器
     *
     * @param clazz         事件类型
     * @param eventListener
     * @param <E>
     */
    <E extends Event> void registerListener(Class<? extends Event> clazz, EventListener<E> eventListener);

    /**
     * 通知所有该事件的监听器
     *
     * @param e
     * @param <E>
     */
    <E extends Event> void notifyListener(E e);

    /**
     * 注册一个事件监听器
     *
     * @param eventClazz 事件类型
     * @param listenerRegistration
     */
    void registerListener(Class<?> eventClazz, EventListenerRegistration listenerRegistration);

    /**
     * 通知所有该事件的监听器
     *
     * @param event
     */
    void notifyListener(Object event);

}

DefaultEventListenerManager增加了一个新的listenerMap用于维护用户自定义的事件注册信息

/**
 * @author 2YSP
 * @date 2022/4/16 
 */
public class DefaultEventListenerManager implements EventListenerManager {
    /**
     * 事件map
     */
    private static Map<Class<? extends Event>, List<EventListener>> map = new HashMap<>(64);
    /**
     * 事件监听器map,key:事件类型
     */
    private static Map<Class<?>, List<EventListenerRegistration>> listenerMap = new HashMap<>(64);


    /**
     * 注册一个事件监听器
     *
     * @param clazz
     * @param eventListener
     * @param <E>
     */
    @Override
    public <E extends Event> void registerListener(Class<? extends Event> clazz, EventListener<E> eventListener) {
        List<EventListener> list = map.get(clazz);
        if (CollectionUtils.isEmpty(list)) {
            map.put(clazz, Lists.newArrayList(eventListener));
        } else {
            list.add(eventListener);
            map.put(clazz, list);
        }
    }

    /**
     * 移除一个事件监听器
     *
     * @param clazz
     * @param <E>
     */
    public <E extends Event> void removeListener(Class<E> clazz) {
        map.remove(clazz);
    }

    /**
     * 通知所有该事件的监听器
     *
     * @param <E>
     */
    @Override
    public <E extends Event> void notifyListener(E e) {
        List<EventListener> eventListeners = map.get(e.getClass());
        if (CollectionUtils.isEmpty(eventListeners)) {
            return;
        }
        eventListeners.forEach(eventListener -> {
            boolean async = false;
            try {
                Method method = eventListener.getClass().getDeclaredMethod(EventConstants.EVENT_METHOD_NAME, Event.class);
                AsyncExecute asyncExecute = AnnotationUtils.findAnnotation(method, AsyncExecute.class);
                async = asyncExecute != null;
            } catch (NoSuchMethodException ex) {
                ex.printStackTrace();
            }
            if (!async) {
                // 同步执行
                eventListener.onEvent(e);
            } else {
                // 异步执行
                EventPoolManager.INSTANCE.execute(() -> eventListener.onEvent(e));
            }
        });
    }


    @Override
    public void registerListener(Class<?> eventClazz, EventListenerRegistration listenerRegistration) {
        if (listenerMap.containsKey(eventClazz)) {
            List<EventListenerRegistration> configList = listenerMap.get(eventClazz);
            configList.add(listenerRegistration);
            listenerMap.put(eventClazz, configList);
        } else {
            listenerMap.put(eventClazz, Lists.newArrayList(listenerRegistration));
        }
    }

    @Override
    public void notifyListener(Object event) {
        Class<?> eventClass = event.getClass();
        List<EventListenerRegistration> eventListenerRegistrations = listenerMap.get(eventClass);
        eventListenerRegistrations.forEach(config -> {
            Class<?> clazz = config.getClazz();
            Object bean = config.getBean();
            Assert.notNull(bean, "the bean of event listener can not be null!");
            Method method = null;
            try {
                method = clazz.getMethod(config.getMethodName(), eventClass);
            } catch (NoSuchMethodException e) {
                e.printStackTrace();
            }
            if (config.getAsync()) {
                Method method2 = method;
                EventPoolManager.INSTANCE.execute(() -> invoke(method2, bean, event));
            } else {
                invoke(method, bean, event);
            }
        });
    }

    private void invoke(Method method, Object target, Object args) {
        try {
            method.invoke(target, args);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }
}

EventListenerRegistration是一个简单的实体类，定义了一些事件监听器注册信息的属性。

package cn.sp.domain;

/**
 * @author Ship
 * @version 1.0.0
 * @description: 事件监听器注册信息
 * @date 2022/04/24 
 */
public class EventListenerRegistration {

    /**
     * 类
     */
    private Class<?> clazz;
    /**
     * 方法名
     */
    private String methodName;
    /**
     * 是否异步执行
     */
    private Boolean async;
    /**
     * 事件监听器对象
     */
    private Object bean;


    public EventListenerRegistration() {
    }

   
    public EventListenerRegistration(Class<?> clazz, String methodName, Boolean async, Object bean) {
        this.clazz = clazz;
        this.methodName = methodName;
        this.async = async;
        this.bean = bean;
    }

   
   // 省略getter/setter方法
}

然后在项目启动时，OnceApplicationContextEventListener分别根据接口和注解初始化监听器配置信息

/**
 * @author Ship
 * @version 1.0.0
 * @description:
 * @date 2022/05/05 
 */
public class OnceApplicationContextEventListener implements ApplicationListener, ApplicationContextAware {

    private static ApplicationContext applicationContext;

    private static EventListenerManager eventListenerManager;

    private static Logger logger = LoggerFactory.getLogger(OnceApplicationContextEventListener.class);

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        OnceApplicationContextEventListener.applicationContext = applicationContext;
    }

    @Override
    public void onApplicationEvent(ApplicationEvent event) {
        if (isOriginalEventSource(event) && event instanceof ApplicationContextEvent) {
            onApplicationContextEvent((ApplicationContextEvent) event);
        }
    }

    private void onApplicationContextEvent(ApplicationContextEvent event) {
        OnceApplicationContextEventListener.eventListenerManager = applicationContext.getBean(EventListenerManager.class);
        logger.info("start to init event spring boot starter config...");
        initConfig();
        logger.info("init event spring boot starter config end.");
    }

    private boolean isOriginalEventSource(ApplicationEvent event) {
        boolean originalEventSource = nullSafeEquals(getApplicationContext(), event.getSource());
        if (!originalEventSource) {
            if (logger.isDebugEnabled()) {
                logger.debug("The source of event[" + event.getSource() + "] is not original!");
            }
        }
        return originalEventSource;
    }

    public ApplicationContext getApplicationContext() {
        if (applicationContext == null) {
            throw new NullPointerException("applicationContext must be not null, it has to invoke " +
                    "setApplicationContext(ApplicationContext) method first if "
                    + ClassUtils.getShortName(getClass()) + " instance is not a Spring Bean");
        }
        return applicationContext;
    }

    /**
     * 初始化配置
     */
    private void initConfig() {
        // 根据接口注册监听器
        registerListenerByInterface();
        // 根据注解注册监听器
        registerListenerByAnnotation();
    }

    private void registerListenerByInterface() {
        Map<String, EventListener> beanMap = applicationContext.getBeansOfType(EventListener.class);
        if (beanMap == null) {
            return;
        }
        beanMap.forEach((key, value) -> {
            // 反射获取onEvent方法的参数类型
            Method[] methods = value.getClass().getDeclaredMethods();
            for (Method method : methods) {
                if (method.getName().equals(EventConstants.EVENT_METHOD_NAME)) {
                    Parameter parameter = method.getParameters()[0];
                    // 参数必须为Event的子类
                    if (parameter.getType().getName().equals(Event.class.getName())) {
                        continue;
                    }
                    eventListenerManager.registerListener((Class<? extends Event>) parameter.getType(), value);
                }
            }
        });
    }

    private void registerListenerByAnnotation() {
        Map<String, Object> map = applicationContext.getBeansWithAnnotation(MyEventListener.class);
        if (map == null) {
            return;
        }
        map.forEach((key, value) -> {
            // 获取所有method
            Class<?> listenerClazz = value.getClass();
            Method[] methods = listenerClazz.getDeclaredMethods();
            for (Method method : methods) {
                MyEventListener myEventListener = AnnotationUtils.findAnnotation(method, MyEventListener.class);
                if (myEventListener == null) {
                    continue;
                }
                Parameter parameter = method.getParameters()[0];
                Class<?> eventClazz = parameter.getType();
                EventListenerRegistration registration = new EventListenerRegistration(listenerClazz, method.getName(),
                        myEventListener.async(), value);
                eventListenerManager.registerListener(eventClazz, registration);
            }
        });
    }
}

上面registerListenerByAnnotation()方法中出现的@MyEventListener注解是为了标记出事件监听器类和方法，为了方便注入Spring容器所以加了@Component元注解，代码如下：

/**
 * @author Ship
 * @version 1.0.0
 * @description: 事件监听标记注解
 * @date 2022/04/24 
 */
@Component
@Documented
@Target(value = {ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface MyEventListener {

    /**
     * 是否异步执行，默认否
     *
     * @return
     */
    boolean async() default false;

}

最后，在自动配置类EventStarterAutoConfigure中注入需要的bean。

@Configuration
public class EventStarterAutoConfigure {

    @Bean
    public EventListenerManager eventListenerManager() {
        return new DefaultEventListenerManager();
    }

    @Bean
    public EventPublisher eventPublisher(@Autowired EventListenerManager eventListenerManager) {
        return new DefaultEventPublisher(eventListenerManager);
    }

    @Bean
    public OnceApplicationContextEventListener onceApplicationListener() {
        return new OnceApplicationContextEventListener();
    }
}

至此核心代码就结束了，还有一个小优化是用枚举来实现线程池的单例模式。

public enum EventPoolManager {

    INSTANCE;

    /**
     * 事件执行线程池
     */
    private final static ExecutorService EVENT_POOL = new ThreadPoolExecutor(4,
            8, 30L, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(512), new ThreadFactoryBuilder().setNameFormat("event-pool-%d").build());

    /**
     * 执行任务
     *
     * @param command
     */
    public void execute(Runnable command) {
        EVENT_POOL.execute(command);
    }
}

三、测试

测试过程比较简单，首先新建测试项目event-spring-boot-starter-sample并引入依赖

 <dependency>
            <groupId>cn.sp</groupId>
            <artifactId>event-spring-boot-starter</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

然后发布事件

@Component
public class Test implements CommandLineRunner {

    @Resource
    private OrderService orderService;

    @Override
    public void run(String... args) throws Exception {
        orderService.create(new Order());
    }
}


@Service
public class OrderService {

    @Resource
    private EventPublisher publisher;


    /**
     * 创建订单
     *
     * @param order
     */
    public void create(Order order) {
        // 发送订单创建事件
        order.setOrderNo("sssss");
//        publisher.publish(new OrderCreateEvent(this, order));
        publisher.publish(order);
    }
}

再监听事件

/**
 * @author Ship
 * @version 1.0.0
 * @description:
 * @date 2022/04/24 
 */
@MyEventListener
public class OrderCreateEventListener3 {

    @MyEventListener(async = true)
    public void onListen(Order order){
        System.out.println(Thread.currentThread().getName() + "--监听订单创建事件3。。。。。。。。。");
    }
}

最后启动项目，控制台输出如下

2022-05-09 22:49:25.804  INFO 7845 --- [           main] .EventSpringBootStarterSampleApplication : Started EventSpringBootStarterSampleApplication in 1.815 seconds (JVM running for 2.789)
event-pool-0--监听订单创建事件3。。。。。。。。。

可以看到OrderCreateEventListener3成功的监听到了创建订单事件。