Spring 事件监听机制及原理分析

作者: 码农奋斗之路 | 来源:发表于2020-11-25 19:59 被阅读0次

简介

在JAVA体系中，有支持实现事件监听机制，在Spring 中也专门提供了一套事件机制的接口，方便我们实现。比如我们可以实现当用户注册后，给他发送一封邮件告诉他注册成功的一些信息，比如用户订阅的主题更新了，通知用户注意及时查看等。

观察者模式

观察者模式还有很多其他的称谓，如发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态上发生变化时，会通知所有观察者对象，使它们能够自动更新自己。

观察者模式一般包含以下几个对象：

Subject：

被观察的对象。它提供一系列方法来增加和删除观察者对象，同时它定义了通知方法notify()。目标类可以是接口，也可以是抽象类或具体类。

ConcreteSubject：

具体的观察对象。Subject的具体实现类，在这里实现通知事件。

Observer：

观察者。这里是抽象的观察者，观察者有一个或者多个。

ConcreteObserver：

具体的观察者。在这里维护观察对象的具体操作。

Java 中的事件机制

Java中提供了基本的事件处理基类：

EventObject：所有事件状态对象都将从其派生的根类；
EventListener：所有事件侦听器接口必须扩展的标记接口；

非常经典的开门案例：

一、创建事件对象

@Getter
@Setter
public class DoorEvent extends EventObject{

    int state;

    public DoorEvent(Object source){
        super(source);
    }
    public DoorEvent(Object source,int state){
        super(source);
        this.state = state;
    }
}

二、事件监听器

public interface DoorListener extends EventListener{
    void doorEvent(DoorEvent doorEvent);
}
public class CloseDoorEvent implements DoorListener{
    @Override
    public void doorEvent(DoorEvent doorEvent){
        if(doorEvent.getState() == -1){
            System.out.println("门关上了");
        }
    }
}
public class OpenDoorListener implements DoorListener{
    @Override
    public void doorEvent(DoorEvent doorEvent){
        if(doorEvent.getState() == 1){
            System.out.println("门打开了");
        }
    }
}

三、测试

public static void main(String[] args){
    List<DoorListener> list = new ArrayList<>();
    list.add(new OpenDoorListener());
    list.add(new CloseDoorEvent());
    for(DoorListener listener : list){
        listener.doorEvent(new DoorEvent(-1,-1));
        listener.doorEvent(new DoorEvent(1,1));
    }
}

四、输出结果

门打开了
门关上了

Spring 中的事件机制

在 Spring 容器中通过ApplicationEven类和 ApplicationListener接口来实现事件监听机制，每次Event 被发布到Spring容器中时都会通知该Listener。需要注意的是，Spring 的事件默认是同步的，调用 publishEvent 方法发布事件后，它会处于阻塞状态，直到Listener接收到事件并处理返回之后才继续执行下去。

代码示例：

一、定义事件对象

@Getter
@Setter
@ToString
public class UserDTO extends ApplicationEvent{
    private Integer userId;
    private String name;
    private Integer age;

    public UserDTO(Object source){
        super(source);
    }
}

二、定义事件监听器，可以通过注解或者实现接口来实现。

@Component
public class UserRegisterSmsListener{

        // 通过注解实现监听器
    @EventListener
    public void handleUserEvent(UserDTO userDTO){
        System.out.println("监听到用户注册，准备发送短信，user:"+userDTO.toString());
    }
}

// 通过实现接口实现监听器
@Component
public class UserRegisterEmailListener implements ApplicationListener<UserDTO>{
    @Override
    public void onApplicationEvent(UserDTO userDTO){
        System.out.println("监听到用户注册，准备发送邮件，user:" + userDTO.toString());
    }
}
@Component
public class UserRegisterMessageListener implements ApplicationListener<UserDTO>{
    @Override
    public void onApplicationEvent(UserDTO userDTO){
        System.out.println("监听到用户注册，给新用户发送首条站内短消息，user:" + userDTO.toString());
    }
}

三、注册服务

public interface UserService{
    void register();
}
@Service
public class UserServiceImpl implements UserService{
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    @Override
    public void register(){
        UserDTO userDTO = new UserDTO(this);
        userDTO.setAge(18);
        userDTO.setName("admol");
        userDTO.setUserId(1001);
        System.out.println("register user");
        eventPublisher.publishEvent(userDTO);
    }
}

四、测试

@Autowired
private UserService userService;

@Test
public void testUserEvent(){
    userService.register();
}

五、输出结果

register user
监听到用户注册，准备发送短信，user：UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，准备发送邮件，user:UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，给新用户发送首条站内短消息，user:UserDTO(userId=1001, name=admol, age=18)

指定监听器的顺序

监听器的发布顺序是按照 bean 自然装载的顺序执行的，Spring 支持两种方式来实现有序

一、实现SmartApplicationListener接口指定顺序。

把上面三个Listener都改成实现SmartApplicationListener接口，并指定getOrder的返回值，返回值越小，优先级越高。

@Component
public class UserRegisterMessageListener implements SmartApplicationListener{

    @Override
    public boolean supportsEventType(Class<? extends ApplicationEvent> eventType){
        return eventType == UserDTO.class;
    }

    @Override
    public boolean supportsSourceType(Class<?> sourceType){
        return true;
    }

    @Override
    public void onApplicationEvent(ApplicationEvent event){
        System.out.println("监听到用户注册，给新用户发送首条站内短消息，user:" + event.toString());
    }

    @Override
    public int getOrder(){
        return -1;
    }
}

另外两个监听器的改造省略，指定改造后的UserRegisterSmsListener返回order为0，UserRegisterEmailListener的getOrder返回1，测试输出结果如下：

register user
监听到用户注册，给新用户发送首条站内短消息，user:UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，准备发送短信，user：UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，准备发送邮件，user:UserDTO(userId=1001, name=admol, age=18)

二、使用注解@Order()

@Component
public class UserRegisterSmsListener{

    @Order(-2)
    @EventListener
    public void handleUserEvent(UserDTO userDTO){
        System.out.println("监听到用户注册，准备发送短信，user:"+userDTO.toString());
    }
}

测试输出结果如下：

register user
监听到用户注册，准备发送短信，user:UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，给新用户发送首条站内短消息，user:UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，准备发送邮件，user:UserDTO(userId=1001, name=admol, age=18)

可以发现，短信监听器最先执行。

异步支持

Spring 事件机制默认是同步阻塞的，如果 ApplicationEventPublisher 发布事件之后他会一直阻塞等待listener 响应，多个 listener 的情况下前面的没有执行完后面的会一直被阻塞。这时候我们可以利用 Spring 提供的线程池注解 @Async 来实现异步线程

一、使用 @Async 之前需要先开启线程池，在启动类上添加 @EnableAsync 注解即可。

@EnableAsync
@SpringBootApplication
public class DemoApplication {
        public static void main(String[] args) {
                SpringApplication.run(DemoApplication.class, args);
        }
}

二、监听器使用异步线程

自定义异步线程池

@Configuration
public class AsyncConfig{

    @Bean("asyncThreadPool")
    public Executor getAsyncExecutor(){
        System.out.println("asyncThreadPool init");
        Executor executor = new ThreadPoolExecutor(
                10,20,60L,TimeUnit.SECONDS
                ,new ArrayBlockingQueue<>(100),new MyThreadFactory());
        return executor;
    }

    class MyThreadFactory implements ThreadFactory{
        final AtomicInteger threadNumber = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r){
            Thread t = new Thread(r);
            t.setName("async-thread-"+threadNumber.getAndIncrement());
            t.setDaemon(true);
            return t;
        }
    }
}

指定监听器的线程池

@Component
public class UserRegisterSmsListener{

    @Order(-2)
    @Async("asyncThreadPool")
    @EventListener
    public void handleUserEvent(UserDTO userDTO){
        System.out.println(Thread.currentThread().getName() + " 监听到用户注册，准备发送短信，user:"+userDTO.toString());
    }
}

三、测试输出结果

register user
监听到用户注册，给新用户发送首条站内短消息，user:UserDTO(userId=1001, name=admol, age=18)
监听到用户注册，准备发送邮件，user:UserDTO(userId=1001, name=admol, age=18)
async-thread-0 监听到用户注册，准备发送短信，user:UserDTO(userId=1001, name=admol, age=18)

Spring事件机制原理分析

Spring事件机制涉及的重要类主要有以下四个：

ApplicationEvent：
事件对象，继承至JDK的类EventObject ，可以携带事件的时间戳

ApplicationListener：
事件监听器，继承至JDK的接口EventListener，该接口被所有的事件监听器实现，比如支持指定顺序的SmartApplicationListener

ApplicationEventMulticaster：
事件管理者，管理监听器和发布事件，ApplicationContext通过委托ApplicationEventMulticaster来发布事件

ApplicationEventPublisher：
事件发布者，该接口封装了事件有关的公共方法，作为ApplicationContext的超级街廓，也是委托 ApplicationEventMulticaster完成事件发布。

源码展示

ApplicationEvent

事件对象ApplicationEvent的主要源代码如下，继承了JAVA的 EventObject 对象：

public abstract class ApplicationEvent extends EventObject {
    private static final long serialVersionUID = 7099057708183571937L;
    private final long timestamp; // 多了一个时间戳属性
    public ApplicationEvent(Object source) {
        super(source);
        this.timestamp = System.currentTimeMillis(); // 初始当前化时间戳
    }
    public final long getTimestamp() {
        return this.timestamp;
    }
}

image

从上面ApplicationEvent的子类关系图种可以发现，ApplicationEvent有一个重要的子类ApplicationContextEvent，而ApplicationContextEvent又有4个重要的子类ContextStartedEvent、ContextRefreshedEvent、ContextClosedEvent、ContextStoppedEvent。

从名字就可以看出，这4个事件都和Spring容器有关系的：

ContextRefreshedEvent：当spring容器context刷新时触发
ContextStartedEvent：当spring容器context启动后触发
ContextStoppedEvent：当spring容器context停止时触发
ContextClosedEvent：当spring容器context关闭时触发，容器被关闭时，其管理的所有单例Bean都被销毁。

当每个事件触发时，相关的监听器就会监听到相应事件，然后触发onApplicationEvent方法。

ApplicationListener

事件监听器，继承DK的接口EventListener

/* ...
 * @author Rod Johnson
 * @author Juergen Hoeller
 * @param <E> the specific ApplicationEvent subclass to listen to
 * @see org.springframework.context.event.ApplicationEventMulticaster
 */
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {

    /**
     * Handle an application event. by jinglingwang.cn
     * @param event the event to respond to
     */
    void onApplicationEvent(E event);

}

注释@param <E> the specific ApplicationEvent subclass to listen to@see ApplicationEventMulticaster 里面说明了事件的广播在ApplicationEventMulticaster类。

ApplicationEventMulticaster

ApplicationEventMulticaster是一个接口，负责管理监听器和发布事件，定义了如下方法：

addApplicationListener(ApplicationListener<?> listener) ：新增一个listener；
addApplicationListenerBean(String listenerBeanName)：新增一个listener，参数为bean name；
removeApplicationListener(ApplicationListener<?> listener)：删除listener；
void removeAllListeners()：删除所有的Listener
removeApplicationListenerBean(String listenerBeanName)：根据bean name 删除listener；
multicastEvent(ApplicationEvent event)：广播事件；
multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType)：广播事件，指定事件的source类型。

AbstractApplicationEventMulticaster 实现了 ApplicationEventMulticaster接口，SimpleApplicationEventMulticaster 继承了AbstractApplicationEventMulticaster ;

AbstractApplicationEventMulticaster 主要实现了管理监听器的方法（上面接口的前5个方法）
SimpleApplicationEventMulticaster 主要实现了事件广播相关的方法（上面接口的最后2个方法）

两个类分别继承了部分上面的方法。

一、先看新增Listener方法实现逻辑：

public abstract class AbstractApplicationEventMulticaster
        implements ApplicationEventMulticaster, BeanClassLoaderAware, BeanFactoryAware {

        private final ListenerRetriever defaultRetriever = new ListenerRetriever(false);

...
        @Override
        public void addApplicationListener(ApplicationListener<?> listener) {
            synchronized (this.retrievalMutex) { // 加排他锁
                // Explicitly remove target for a proxy, if registered already,
                // in order to avoid double invocations of the same listener.
                Object singletonTarget = AopProxyUtils.getSingletonTarget(listener);
                if (singletonTarget instanceof ApplicationListener) {
          // 删除，避免重复调用
                    this.defaultRetriever.applicationListeners.remove(singletonTarget);
                }
        // 加入到Set LinkedHashSet 集合中
                this.defaultRetriever.applicationListeners.add(listener);
                this.retrieverCache.clear(); // 缓存
            }
        }
...
}

最核心的一句代码：this.defaultRetriever.applicationListeners.add(listener);

ListenerRetriever类是AbstractApplicationEventMulticaster类的内部类，里面有两个集合，用来记录维护事件监听器。

private class ListenerRetriever {

        public final Set<ApplicationListener<?>> applicationListeners = new LinkedHashSet<>();
        public final Set<String> applicationListenerBeans = new LinkedHashSet<>();
        ...
}

这就和设计模式中的发布订阅模式一样了，维护一个List，用来管理所有的订阅者，当发布者发布消息时，遍历对应的订阅者列表，执行各自的回调handler。

二、看SimpleApplicationEventMulticaster类实现的广播事件逻辑：

@Override
public void multicastEvent(ApplicationEvent event) {
    multicastEvent(event, resolveDefaultEventType(event)); // 继续调用下面的广播方法
}

@Override
public void multicastEvent(final ApplicationEvent event, ResolvableType eventType) {
    ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
  // 遍历监听器列表
    for (final ApplicationListener<?> listener : getApplicationListeners(event, type)) {
        Executor executor = getTaskExecutor();
        if (executor != null) { // 是否指定了线程池
            executor.execute(new Runnable() {
                @Override
                public void run() { // 线程池执行
                    invokeListener(listener, event);
                }
            });
        }
        else { // 普通执行
            invokeListener(listener, event);
        }
    }
}

代码分析：

首先根据事件类型，获取事件监听器列表：getApplicationListeners(event, type)
遍历监听器列表，for循环
判断是否有线程池，如果有，在线程池执行
否则直接执行

我们再看看 invokeListener方法的逻辑：

protected void invokeListener(ApplicationListener<?> listener, ApplicationEvent event) {
        ErrorHandler errorHandler = getErrorHandler();
        if (errorHandler != null) { // 是否有错误处理
            try {
                doInvokeListener(listener, event);
            } catch (Throwable err) {
                errorHandler.handleError(err);
            }
        } else {
            doInvokeListener(listener, event); // 直接执行
        }
    }

核心逻辑就是继续调用doInvokeListener方法：

private void doInvokeListener(ApplicationListener listener, ApplicationEvent event) {
        try {
            listener.onApplicationEvent(event);// 执行监听器事件
        }
        catch (ClassCastException ex) {
            String msg = ex.getMessage();
            if (msg == null || msg.startsWith(event.getClass().getName())) {
                // Possibly a lambda-defined listener which we could not resolve the generic event type for
                Log logger = LogFactory.getLog(getClass());
                if (logger.isDebugEnabled()) {
                    logger.debug("Non-matching event type for listener: " + listener, ex);
                }
            }
            else {
                throw ex;
            }
        }
    }

发现最后实际就是调用的 listener.onApplicationEvent(event); 也就是我们通过实现接口ApplicationListener的方式来实现监听器的onApplicationEvent实现逻辑。

ApplicationEventPublisher类

在我们的发布事件逻辑代码的地方，通过查看 eventPublisher.publishEvent(userDTO);方法可以发现ApplicationEventPublisher是一个接口，publishEvent方法的逻辑实现主要在类AbstractApplicationContext中：

public abstract class AbstractApplicationContext extends DefaultResourceLoader
        implements ConfigurableApplicationContext, DisposableBean {
...
        private Set<ApplicationEvent> earlyApplicationEvents;
...
        @Override
        public void publishEvent(ApplicationEvent event) {
            publishEvent(event, null); // 调用下面的方法
        }
    // 发布事件主要逻辑
        protected void publishEvent(Object event, ResolvableType eventType) {
                Assert.notNull(event, "Event must not be null");
                if (logger.isTraceEnabled()) {
                    logger.trace("Publishing event in " + getDisplayName() + ": " + event);
                }

                // 事件装饰为 ApplicationEvent
                ApplicationEvent applicationEvent;
                if (event instanceof ApplicationEvent) {
                    applicationEvent = (ApplicationEvent) event;
                } else {
                    applicationEvent = new PayloadApplicationEvent<Object>(this, event);
                    if (eventType == null) {
                        eventType = ((PayloadApplicationEvent) applicationEvent).getResolvableType();
                    }
                }

                // 容器启动的时候 earlyApplicationEvents 可能还没有初始化
                if (this.earlyApplicationEvents != null) {
                    this.earlyApplicationEvents.add(applicationEvent); // 加入到集合，同一广播
                } else {
          // 还没初始化，直接广播事件
                    getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
                }

                // 通过父上下文发布事件.
                if (this.parent != null) {
                    if (this.parent instanceof AbstractApplicationContext) {
                        ((AbstractApplicationContext) this.parent).publishEvent(event, eventType);
                    }
                    else {
                        this.parent.publishEvent(event);
                    }
                }
            }
...
}

这段代码的主要逻辑在这：

if (this.earlyApplicationEvents != null) {
    this.earlyApplicationEvents.add(applicationEvent);
}
else {
    getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
}

可以发现earlyApplicationEvents也是一个Set集合，如果这个集合已经初始化了，就把事件加入到集合中，否则直接调用multicastEvent执行事件监听逻辑。

我们跟踪找到初始化这个集合的地方，发现在方法protected void prepareRefresh()中：

protected void prepareRefresh() {
        this.startupDate = System.currentTimeMillis();
        this.closed.set(false);
        this.active.set(true);

        if (logger.isInfoEnabled()) {
            logger.info("Refreshing " + this);
        }

        initPropertySources();

        getEnvironment().validateRequiredProperties();

        **this.earlyApplicationEvents = new LinkedHashSet<ApplicationEvent>();**
    }

继续跟踪调用这个方法的地方，发现在AbstractApplicationContext.refresh()方法中，而这个方法是Spring容器初始化必须要调用的过程，非常的重要。

那在什么地方使用到了这个集合呢？我们继续跟踪发现在 protected void registerListeners() 方法中，代码如下：

protected void registerListeners() {
        // Register statically specified listeners first.
        for (ApplicationListener<?> listener : getApplicationListeners()) {
            getApplicationEventMulticaster().addApplicationListener(listener);
        }

        // Do not initialize FactoryBeans here: We need to leave all regular beans
        // uninitialized to let post-processors apply to them! jinglingwang.cn
        String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false);
        for (String listenerBeanName : listenerBeanNames) {
            getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName);
        }

        // 拿到集合引用
        Set<ApplicationEvent> ****earlyEventsToProcess = this.earlyApplicationEvents;
        this.earlyApplicationEvents = null; // 把之前的集合置为null
        if (earlyEventsToProcess != null) { // 如果集合不为空，则广播里面的事件
            for (ApplicationEvent earlyEvent : earlyEventsToProcess) {
                getApplicationEventMulticaster().multicastEvent(earlyEvent);
            }
        }
    }

逻辑是先获得该集合的引用，然后置空之前的集合，然后遍历集合，进行广播事件multicastEvent，这个方法的逻辑上面已经说过了。

而registerListeners这个方法是在什么时候调用的呢？通过跟踪发现也是在AbstractApplicationContext.refresh()方法中。

只不过基本是在方法逻辑的最后，也就是Spring已经容器初始化完成了。

@Override
    public void refresh() throws BeansException, IllegalStateException {
        synchronized (this.startupShutdownMonitor) {
            // Prepare this context for refreshing.
            **prepareRefresh**();

            ....
            try {
                onRefresh();

                // Check for listener beans and register them.
                **registerListeners**();

                // Instantiate all remaining (non-lazy-init) singletons.
                finishBeanFactoryInitialization(beanFactory);

                // Last step: publish corresponding event.
                **finishRefresh**();
            }

            catch (BeansException ex) {
            ...
            }

            finally {
                ...
            }
        }
    }

容器初始化之前和之后都有可能进行广播事件。

总结

事件监听机制和观察者模式非常相似
JDK 也有实现提供事件监听机制
Spring 的事件机制也是基于JDK 来扩展的
Spring 的事件机制默认是同步阻塞的
Spring 容器初始化前后都可能进行广播事件

Spring 事件监听机制及原理分析

简介

观察者模式

Java 中的事件机制

Spring 中的事件机制

指定监听器的顺序

异步支持

Spring事件机制原理分析

源码展示

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 杂谈

java高级开发群

Spring-Boot

开发技巧

Java学习笔记