ApplicationListener事件监听、自定义事件、异步执行

淩亂°似流年 2022-11-01 00:52 408阅读 0赞

ApplicationEvent以及Listener是Spring为我们提供的一个事件监听、订阅的实现，内部实现原理是观察者设计模式，设计初衷也是为了系统业务逻辑之间的解耦，提高可扩展性以及可维护性。事件发布者并不需要考虑谁去监听，监听具体的实现内容是什么，发布者的工作只是为了发布事件而已。

Spring提供的内置事件：

*  ContextRefreshedEvent
 *  ContextStartedEvent
 *  ContextStoppedEvent
 *  ContextClosedEvent

# 如何使用 #

## 监听容器的刷新与关闭事件 ##

自定义一个ApplicationListener：

package com.morris.spring.listener;
    
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    
    public class ContextRefreshedListenerimplements ApplicationListener<ContextRefreshedEvent> { 
    
    	@Override
    	public void onApplicationEvent(ContextRefreshedEvent event) { 
    		System.out.println("context refresh");
    	}
    }

注入到容器中：

package com.morris.spring.demo.annotation;
    
    import com.morris.spring.listener.ContextRefreshedApplicationListener;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    
    public class ContextRefreshedApplicationListenerDemo { 
    	public static void main(String[] args) { 
    		AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
    		applicationContext.register(ContextRefreshedListener.class);
    		applicationContext.refresh();
    	}
    }

## 自定义监听事件 ##

自定义的监听事件需要继承ApplicationEvent：

package com.morris.spring.event;
    
    import org.springframework.context.ApplicationEvent;
    
    public class CustomEvent extends ApplicationEvent { 
    
    	private static final long serialVersionUID = 1L;
    
    	public CustomEvent(Object source) { 
    		super(source);
    	}
    }

监听的时候使用ApplicationEvent的子类CustomEvent：

package com.morris.spring.listener;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    
    public class CustomEventListener implements ApplicationListener<CustomEvent> { 
    
    	@Override
    	public void onApplicationEvent(CustomEvent event) { 
    		System.out.println("custom event: " + event.getSource());
    	}
    }

可以使用AnnotationConfigApplicationContext发布事件：

package com.morris.spring.demo.annotation;
    
    import com.morris.spring.event.CustomEvent;
    import com.morris.spring.listener.CustomEventListener;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    
    public class CustomEventListenerDemo { 
    	public static void main(String[] args) { 
    		AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
    		applicationContext.register(CustomEventListener.class);
    		applicationContext.refresh();
    
    		applicationContext.publishEvent(new CustomEvent("自定义事件"));
    	}
    }

可以向bean中注入一个ApplicationEventPublisher来发布事件：

package com.morris.spring.service;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationEventPublisher;
    
    public class CustomEventService { 
    
    	@Autowired
    	private ApplicationEventPublisher applicationEventPublisher;
    
    	public void publishEvent() { 
    		applicationEventPublisher.publishEvent(new CustomEvent("自定义事件"));
    	}
    }

可以通过实现ApplicationEventPublisherAware接口注入ApplicationEventPublisher来发布事件：

package com.morris.spring.service;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.context.ApplicationEventPublisher;
    import org.springframework.context.ApplicationEventPublisherAware;
    
    public class CustomEventService2 implements ApplicationEventPublisherAware { 
    
    	private ApplicationEventPublisher applicationEventPublisher;
    
    	public void publishEvent() { 
    		applicationEventPublisher.publishEvent(new CustomEvent("自定义事件"));
    	}
    
    	@Override
    	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { 
    		this.applicationEventPublisher = applicationEventPublisher;
    	}
    }

由于ApplicationContext实现了ApplicationEventPublisher接口，也可以直接注入ApplicationContext来发布事件。

## 使用@EventListener监听事件 ##

由于实现ApplicationListener接口有很大的侵入性，可以使用@EventListener注解随时随地监听事件，这样一个Service中可以监听多个事件：

package com.morris.spring.listener;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.context.event.EventListener;
    
    public class CustomEventListener2 { 
    
    	@EventListener
    	public void listenContextRefreshedEvent(ContextRefreshedEvent event) { 
    		System.out.println("context refresh");
    	}
    
    	@EventListener
    	public void listenCustomEvent(CustomEvent event) { 
    		System.out.println("custom event: " + event.getSource());
    	}
    
    }

还可以指定监听的事件类型：

package com.morris.spring.listener;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.context.ApplicationEvent;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.context.event.EventListener;
    
    public class CustomEventListener3 { 
    
    	@EventListener({ ContextRefreshedEvent.class, CustomEvent.class})
    	public void listenEvent(ApplicationEvent event) { 
    		System.out.println(event);
    	}
    }

## 异步发送消息 ##

消息的发送默认都是同步的，如果要异步发送消息，首先要在配置类上开启异步功能@EnableAsync：

package com.morris.spring.config;
    
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    
    @Configuration
    @EnableAsync // 开启异步
    public class EventListenerConfig { 
    }

在监听的方法上加上@Async：

package com.morris.spring.listener;
    
    import com.morris.spring.event.CustomEvent;
    import org.springframework.context.ApplicationEvent;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.context.event.EventListener;
    import org.springframework.scheduling.annotation.Async;
    
    public class AsyncCustomEventListener { 
    	@EventListener({ ContextRefreshedEvent.class, CustomEvent.class})
    	@Async // 异步
    	public void listenEvent(ApplicationEvent event) { 
    		System.out.println(Thread.currentThread().getName());
    		System.out.println(event);
    	}
    }

也可以自定义执行异步消息的线程池（默认就是SimpleAsyncTaskExecutor）：

@Bean
    public TaskExecutor executor() { 
    	return new SimpleAsyncTaskExecutor("eventListen-");
    }

# ApplicationListener原理分析 #

### 发布消息 ###

发布消息入口：

// org.springframework.context.support.AbstractApplicationContext#publishEvent(org.springframework.context.ApplicationEvent)
    public void publishEvent(ApplicationEvent event) { 
        publishEvent(event, null);
    }
    
    protected void publishEvent(Object event, @Nullable ResolvableType eventType) { 
    ...
            /** * @see SimpleApplicationEventMulticaster#multicastEvent(org.springframework.context.ApplicationEvent, org.springframework.core.ResolvableType) */
            // 发布消息
            getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
        }
    ... ...
    }

然后调用SimpleApplicationEventMulticaster来进行广播消息：

// org.springframework.context.event.SimpleApplicationEventMulticaster#multicastEvent(org.springframework.context.ApplicationEvent, org.springframework.core.ResolvableType)
    public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) { 
        ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
        // 如果有线程池，将会异步执行
        Executor executor = getTaskExecutor();
        for (ApplicationListener<?> listener : getApplicationListeners(event, type)) { 
            if (executor != null) { 
                executor.execute(() -> invokeListener(listener, event));
            }
            else { 
                invokeListener(listener, event);
            }
        }
    }
    
    protected void invokeListener(ApplicationListener<?> listener, ApplicationEvent event) { 
        ErrorHandler errorHandler = getErrorHandler();
        if (errorHandler != null) { 
            try { 
                doInvokeListener(listener, event);
            }
            catch (Throwable err) { 
                errorHandler.handleError(err);
            }
        }
        else { 
            // ApplicationListener.调用onApplicationEvent
            doInvokeListener(listener, event);
        }
    }
    
    private void doInvokeListener(ApplicationListener listener, ApplicationEvent event) { 
        try { 
            listener.onApplicationEvent(event);
        }
        catch (ClassCastException ex) { 
            String msg = ex.getMessage();
            if (msg == null || matchesClassCastMessage(msg, event.getClass())) { 
                // Possibly a lambda-defined listener which we could not resolve the generic event type for
                // -> let's suppress the exception and just log a debug message.
                Log logger = LogFactory.getLog(getClass());
                if (logger.isTraceEnabled()) { 
                    logger.trace("Non-matching event type for listener: " + listener, ex);
                }
            }
            else { 
                throw ex;
            }
        }
    }

### 何时注入SimpleApplicationEventMulticaster ###

从上面的源码可以发现spring是通过SimpleApplicationEventMulticaster事件多播器来发布消息的，那么这个类是何时注入的呢？

// org.springframework.context.support.AbstractApplicationContext#initApplicationEventMulticaster
    protected void initApplicationEventMulticaster() { 
    	ConfigurableListableBeanFactory beanFactory = getBeanFactory();
    	if (beanFactory.containsLocalBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME)) { 
    		this.applicationEventMulticaster =
    				beanFactory.getBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class);
    		if (logger.isTraceEnabled()) { 
    			logger.trace("Using ApplicationEventMulticaster [" + this.applicationEventMulticaster + "]");
    		}
    	}
    	else { 
    		// 直接new，然后放入到spring一级缓存中
    		this.applicationEventMulticaster = new SimpleApplicationEventMulticaster(beanFactory);
    		beanFactory.registerSingleton(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, this.applicationEventMulticaster);
    		if (logger.isTraceEnabled()) { 
    			logger.trace("No '" + APPLICATION_EVENT_MULTICASTER_BEAN_NAME + "' bean, using " +
    					"[" + this.applicationEventMulticaster.getClass().getSimpleName() + "]");
    		}
    	}
    }

### 何时注入ApplicationListener ###

spring在发布消息时，会从SimpleApplicationEventMulticaster中拿出所有的ApplicationListener，那么这些ApplicationListener何时被注入的呢？

// org.springframework.context.support.AbstractApplicationContext#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!
    	String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false);
    	for (String listenerBeanName : listenerBeanNames) { 
    		// 添加到SimpleApplicationEventMulticaster中
    		getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName);
    	}
    
    	// Publish early application events now that we finally have a multicaster...
    	Set<ApplicationEvent> earlyEventsToProcess = this.earlyApplicationEvents;
    	this.earlyApplicationEvents = null;
    	if (!CollectionUtils.isEmpty(earlyEventsToProcess)) { 
    		for (ApplicationEvent earlyEvent : earlyEventsToProcess) { 
    			getApplicationEventMulticaster().multicastEvent(earlyEvent);
    		}
    	}
    }

##  ##

EventListenerMethodProcessor何时被注入

### @EventListener的原理 ###

@EventListener的功能是通过EventListenerMethodProcessor来实现的，在注解扫描时被注AnnotationConfigUtils\#registerAnnotationConfigProcessors()。

EventListenerMethodProcessor主要实现了两个接口：SmartInitializingSingleton和BeanFactoryPostProcessor

先来看看BeanFactoryPostProcessor的postProcessBeanFactory()，这个方法主要是保存beanFactory和eventListenerFactories，后面的方法将会使用到：

public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) { 
        // 保存beanFactory
        this.beanFactory = beanFactory;
    
        /** * EventListenerFactory[DefaultEventListenerFactory]在何处被注入？ * @see AnnotationConfigUtils#registerAnnotationConfigProcessors(org.springframework.beans.factory.support.BeanDefinitionRegistry, java.lang.Object) */
        Map<String, EventListenerFactory> beans = beanFactory.getBeansOfType(EventListenerFactory.class, false, false);
        List<EventListenerFactory> factories = new ArrayList<>(beans.values());
        AnnotationAwareOrderComparator.sort(factories);
        // 保存eventListenerFactories
        this.eventListenerFactories = factories;
    }

再来看看SmartInitializingSingleton的afterSingletonsInstantiated()方法，这个方法会在所有的bean初始化完后执行。

public void afterSingletonsInstantiated() { 
        ConfigurableListableBeanFactory beanFactory = this.beanFactory;
        Assert.state(this.beanFactory != null, "No ConfigurableListableBeanFactory set");
        String[] beanNames = beanFactory.getBeanNamesForType(Object.class);
        for (String beanName : beanNames) { 
            if (!ScopedProxyUtils.isScopedTarget(beanName)) { 
                Class<?> type = null;
                try { 
                    type = AutoProxyUtils.determineTargetClass(beanFactory, beanName);
                }
                catch (Throwable ex) { 
                    // An unresolvable bean type, probably from a lazy bean - let's ignore it.
                    if (logger.isDebugEnabled()) { 
                        logger.debug("Could not resolve target class for bean with name '" + beanName + "'", ex);
                    }
                }
                if (type != null) { 
                    if (ScopedObject.class.isAssignableFrom(type)) { 
                        try { 
                            Class<?> targetClass = AutoProxyUtils.determineTargetClass(
                                    beanFactory, ScopedProxyUtils.getTargetBeanName(beanName));
                            if (targetClass != null) { 
                                type = targetClass;
                            }
                        }
                        catch (Throwable ex) { 
                            // An invalid scoped proxy arrangement - let's ignore it.
                            if (logger.isDebugEnabled()) { 
                                logger.debug("Could not resolve target bean for scoped proxy '" + beanName + "'", ex);
                            }
                        }
                    }
                    try { 
                        processBean(beanName, type);
                    }
                    catch (Throwable ex) { 
                        throw new BeanInitializationException("Failed to process @EventListener " +
                                "annotation on bean with name '" + beanName + "'", ex);
                    }
                }
            }
        }
    }
    
    private void processBean(final String beanName, final Class<?> targetType) { 
        if (!this.nonAnnotatedClasses.contains(targetType) &&
                !targetType.getName().startsWith("java") &&
                !isSpringContainerClass(targetType)) { 
    
            Map<Method, EventListener> annotatedMethods = null;
            try { 
                annotatedMethods = MethodIntrospector.selectMethods(targetType,
                        (MethodIntrospector.MetadataLookup<EventListener>) method ->
                                AnnotatedElementUtils.findMergedAnnotation(method, EventListener.class));
            }
            catch (Throwable ex) { 
                // An unresolvable type in a method signature, probably from a lazy bean - let's ignore it.
                if (logger.isDebugEnabled()) { 
                    logger.debug("Could not resolve methods for bean with name '" + beanName + "'", ex);
                }
            }
    
            if (CollectionUtils.isEmpty(annotatedMethods)) { 
                this.nonAnnotatedClasses.add(targetType);
                if (logger.isTraceEnabled()) { 
                    logger.trace("No @EventListener annotations found on bean class: " + targetType.getName());
                }
            }
            else { 
                // Non-empty set of methods
                ConfigurableApplicationContext context = this.applicationContext;
                Assert.state(context != null, "No ApplicationContext set");
                List<EventListenerFactory> factories = this.eventListenerFactories;
                Assert.state(factories != null, "EventListenerFactory List not initialized");
                for (Method method : annotatedMethods.keySet()) { 
                    for (EventListenerFactory factory : factories) { 
                        if (factory.supportsMethod(method)) { 
                            Method methodToUse = AopUtils.selectInvocableMethod(method, context.getType(beanName));
                            // 使用factory创建一个ApplicationListener
                            ApplicationListener<?> applicationListener =
                                    factory.createApplicationListener(beanName, targetType, methodToUse);
                            if (applicationListener instanceof ApplicationListenerMethodAdapter) { 
                                ((ApplicationListenerMethodAdapter) applicationListener).init(context, this.evaluator);
                            }
                            // 添加至容器中
                            context.addApplicationListener(applicationListener);
                            break;
                        }
                    }
                }
                if (logger.isDebugEnabled()) { 
                    logger.debug(annotatedMethods.size() + " @EventListener methods processed on bean '" +
                            beanName + "': " + annotatedMethods);
                }
            }
        }
    }