Spring事件监听源码分析
众所周知，在Spring中集成了内部的消息分发机制，可以在代码中使用注解@EventListener在方法上或者实现ApplicationListener接口实现对事件消息的监听，当有对应事件push到ApplicationEventPublisher中，监听者可以接收到对应的消息。
ApplicationContext实现了ApplicationEventPublisher接口，调用ApplicationEventPublisher即可发布事件消息。ApplicationEventPublisher中的pushEvent方法在AbstractApplicationContext中实现，下面我们对AbstractApplicationContext中的pushEvent源码进行解析。
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);
}
// Decorate event as an ApplicationEvent if necessary
ApplicationEvent applicationEvent;
//如果event是ApplicationEvent，则applicationEvent=event
if (event instanceof ApplicationEvent) {
applicationEvent = (ApplicationEvent) event;
}
//如果不是，则applicationEvent=new PayloadApplicationEvent，将event作为payload,source = this
else {
applicationEvent = new PayloadApplicationEvent<Object>(this, event);
if (eventType == null) {
eventType = ((PayloadApplicationEvent) applicationEvent).getResolvableType();
}
}
// Multicast right now if possible - or lazily once the multicaster is initialized
if (this.earlyApplicationEvents != null) {
this.earlyApplicationEvents.add(applicationEvent);
}
else {
//对事件进行广播，ApplicationEventMulticaster目前只有SimpleApplicationEventMulticaster一个实现类
getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
}
// Publish event via parent context as well...
if (this.parent != null) {
if (this.parent instanceof AbstractApplicationContext) {
((AbstractApplicationContext) this.parent).publishEvent(event, eventType);
}
else {
this.parent.publishEvent(event);
}
}
}
下面对SimpleApplicationEventMulticaster进行分析。
@Override
public void multicastEvent(final ApplicationEvent event, ResolvableType eventType) {
//如果eventType==null，则调用resolveDefaultEventType方法进行解析事件类型，实际调用了ResolvableType.forInstance方法
ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
//获取ApplicationListener
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);
}
}
}
//ResolvableType.forInstance
public static ResolvableType forInstance(Object instance) {
Assert.notNull(instance, "Instance must not be null");
//实际上ResolvableTypeProvider的实现类只有PayloadApplicationEvent一个
//如果instance（event）是PayloadApplicationEvent
if (instance instanceof ResolvableTypeProvider) {
ResolvableType type = ((ResolvableTypeProvider) instance).getResolvableType();
if (type != null) {
return type;
}
}
//否则以instance的class作为ResolvableType
return ResolvableType.forClass(instance.getClass());
}
protected Collection<ApplicationListener<?>> getApplicationListeners(
ApplicationEvent event, ResolvableType eventType) {
//获取事件源
Object source = event.getSource();
Class<?> sourceType = (source != null ? source.getClass() : null);
ListenerCacheKey cacheKey = new ListenerCacheKey(eventType, sourceType);
// Quick check for existing entry on ConcurrentHashMap...
//在缓存中查询
ListenerRetriever retriever = this.retrieverCache.get(cacheKey);
if (retriever != null) {
return retriever.getApplicationListeners();
}
if (this.beanClassLoader == null ||
(ClassUtils.isCacheSafe(event.getClass(), this.beanClassLoader) &&
(sourceType == null || ClassUtils.isCacheSafe(sourceType, this.beanClassLoader)))) {
// Fully synchronized building and caching of a ListenerRetriever
synchronized (this.retrievalMutex) {
retriever = this.retrieverCache.get(cacheKey);
if (retriever != null) {
return retriever.getApplicationListeners();
}
retriever = new ListenerRetriever(true);
Collection<ApplicationListener<?>> listeners =
retrieveApplicationListeners(eventType, sourceType, retriever);
//放入缓存
this.retrieverCache.put(cacheKey, retriever);
return listeners;
}
}
else {
// No ListenerRetriever caching -> no synchronization necessary
return retrieveApplicationListeners(eventType, sourceType, null);
}
}
/**
* Actually retrieve the application listeners for the given event and source type.
* @param eventType the event type
* @param sourceType the event source type
* @param retriever the ListenerRetriever, if supposed to populate one (for caching purposes)
* @return the pre-filtered list of application listeners for the given event and source type
*/
private Collection<ApplicationListener<?>> retrieveApplicationListeners(
ResolvableType eventType, Class<?> sourceType, ListenerRetriever retriever) {
LinkedList<ApplicationListener<?>> allListeners = new LinkedList<ApplicationListener<?>>();
Set<ApplicationListener<?>> listeners;
Set<String> listenerBeans;
synchronized (this.retrievalMutex) {
//载入默认的监听器
listeners = new LinkedHashSet<ApplicationListener<?>>(this.defaultRetriever.applicationListeners);
//载入默认监听器Bean's name
listenerBeans = new LinkedHashSet<String>(this.defaultRetriever.applicationListenerBeans);
}
for (ApplicationListener<?> listener : listeners) {
//如果监听器支持该eventType和sourceType
if (supportsEvent(listener, eventType, sourceType)) {
if (retriever != null) {
retriever.applicationListeners.add(listener);
}
allListeners.add(listener);
}
}
if (!listenerBeans.isEmpty()) {
BeanFactory beanFactory = getBeanFactory();
for (String listenerBeanName : listenerBeans) {
try {
Class<?> listenerType = beanFactory.getType(listenerBeanName);
//如果监听器支持该eventType和sourceType
if (listenerType == null || supportsEvent(listenerType, eventType)) {
ApplicationListener<?> listener =
beanFactory.getBean(listenerBeanName, ApplicationListener.class);
if (!allListeners.contains(listener) && supportsEvent(listener, eventType, sourceType)) {
if (retriever != null) {
retriever.applicationListenerBeans.add(listenerBeanName);
}
allListeners.add(listener);
}
}
}
catch (NoSuchBeanDefinitionException ex) {
// Singleton listener instance (without backing bean definition) disappeared -
// probably in the middle of the destruction phase
}
}
}
AnnotationAwareOrderComparator.sort(allListeners);
return allListeners;
}
我们在使用Spring提供的事件发布/监听时，很少会实现ApplicationListener接口，更多的是在方法上面使用@EventListener注解实现对某种事件的监听，那@EventListener注解的方法是如何变成ApplicationListener的呢？其实是通过
EventListenerMethodProcessor对@EventListener注解的方法的方法进行处理的，EventListenerMethodProcessor对SmartInitializingSingleton接口进行了实现，该接口是当BeanFactory初始化完成后，BeanFactory会调用该接口的afterSingletonsInstantiated方法进行一些操作。下面对EventListenerMethodProcessor的核心代码进行分析。
@Override
public void afterSingletonsInstantiated() {
//从BeanFactory中获取EventListenerFactory，EventListenerFactory共有2个实现，一个是DefaultEventListenerFactory，对普通的@EventListener进行解析，另一个是TransactionalEventListenerFactory，可以对@TransactionalEventListener进行解析。
List<EventListenerFactory> factories = getEventListenerFactories();
//获取所有的BeanName
String[] beanNames = this.applicationContext.getBeanNamesForType(Object.class);
//对所有Bean遍历
for (String beanName : beanNames) {
if (!ScopedProxyUtils.isScopedTarget(beanName)) {
Class<?> type = null;
try {
type = AutoProxyUtils.determineTargetClass(this.applicationContext.getBeanFactory(), 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 {
type = AutoProxyUtils.determineTargetClass(this.applicationContext.getBeanFactory(),
ScopedProxyUtils.getTargetBeanName(beanName));
}
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(factories, beanName, type);
}
catch (Throwable ex) {
throw new BeanInitializationException("Failed to process @EventListener " +
"annotation on bean with name '" + beanName + "'", ex);
}
}
}
}
}
protected void processBean(final List<EventListenerFactory> factories, final String beanName, final Class<?> targetType) {
if (!this.nonAnnotatedClasses.contains(targetType)) {
Map<Method, EventListener> annotatedMethods = null;
try {
//查找带@EventListener注解的方法
annotatedMethods = MethodIntrospector.selectMethods(targetType,
new MethodIntrospector.MetadataLookup<EventListener>() {
@Override
public EventListener inspect(Method method) {
return 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);
}
}
//如果没有找到带EventListener注解的方法
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
//遍历方法
for (Method method : annotatedMethods.keySet()) {
for (EventListenerFactory factory : factories) {
//EventListenerFactory是否支持对该method的处理
if (factory.supportsMethod(method)) {
Method methodToUse = AopUtils.selectInvocableMethod(
method, this.applicationContext.getType(beanName));
//使用EventListenerFactory创建ApplicationListener，两种EventListenerFactory创建的ApplicationListener其实都是ApplicationListenerMethodAdapter
ApplicationListener<?> applicationListener =
factory.createApplicationListener(beanName, targetType, methodToUse);
//如果applicationListener是ApplicationListenerMethodAdapter，则调用砌init方法
if (applicationListener instanceof ApplicationListenerMethodAdapter) {
((ApplicationListenerMethodAdapter) applicationListener)
.init(this.applicationContext, this.evaluator);
}
//向application中注册applicationListener
this.applicationContext.addApplicationListener(applicationListener);
break;
}
}
}
if (logger.isDebugEnabled()) {
logger.debug(annotatedMethods.size() + " @EventListener methods processed on bean '" +
beanName + "': " + annotatedMethods);
}
}
}
}
