图文Debug深入研究下Spring Boot 事件发布监听源码

vlambda
2022-04-26

图文Debug深入研究下Spring Boot 事件发布监听源码

Spring Boot 版本

2.6

使用

定义事件

public class ZouEvent extends ApplicationEvent {

    @Getter
    private final String msg ;


    public ZouEvent(Object source, String msg) {
        super(source);
        this.msg = msg;
    }
}

定义事件订阅者

这里spring 提供两种方式订阅

1. 实现`ApplicationListener` 接口

@Component
public class ZouListener implements ApplicationListener<ZouEvent> {
    
    @Override
    public void onApplicationEvent(ZouEvent event) {
        System.out.println("实现接口接受事件ZouEvent");
        System.out.println(event.getMsg());
    }
}

使用 @EventListener注解

@Component
public class ZouListenerByAnnotation {

    @EventListener
    public void onApplicationEvent(ZouEvent event) {
        System.out.println("接受事件ZouEvent");
        System.out.println(event.getMsg());
    }

}

测试

/**
 * @author : wh
 * @date : 2022/4/11 16:35
 * @description:
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ZouApplication.class)
public class ZouTest {

    @Autowired
    ApplicationContext applicationContext;
    
    @Test
    public void test() {
        ZouEvent zouEvent = new ZouEvent(this, "测试msg");
        applicationContext.publishEvent(zouEvent);
    }
}

可以看到是可以正常注册和消费事件的

源码分析

事件发布

源码入库很简单，我们直接在

applicationContext.publishEvent(zouEvent);

处打个断点，老规矩，一步一步debug

首先可以看到的是

ApplicationContext实现类其实使用的是GenericApplicationContext

然后我们简单看看GenericApplicationContextUML图

整体还是挺复杂的，但是我们看看我们今天的重点，其中方法publishEvent其实是继承抽象类AbstractApplicationContext的方法,我们可以debug看看

这里没什么重要的信息，我们需要继续深入,继续看AbstractApplicationContext的publishEvent方法

protected void publishEvent(Object event, @Nullable ResolvableType eventType) {
  Assert.notNull(event, "Event must not be null");

  // Decorate event as an ApplicationEvent if necessary
  ApplicationEvent applicationEvent;
  if (event instanceof ApplicationEvent) {
   applicationEvent = (ApplicationEvent) event;
  }
  else {
   applicationEvent = new PayloadApplicationEvent<>(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 {
   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);
   }
  }
 }

直接判断事件 event是否为 ApplicationEvent，是则直强转为 ApplicationEvent
通过方法 getApplicationEventMulticaster获取 ApplicationEventMulticaster

getApplicationEventMulticaster()

可以看到这里获取的ApplicationEventMulticaster实际就是SimpleApplicationEventMulticaster，目前来看接口ApplicationEventMulticaster 的实现类就这一个，如果想看哪里初始化的我们跟踪源码不难发现就是在initApplicationEventMulticaster方法初始化的，具体这里就不细说了图文Debug深入研究下Spring Boot 事件发布监听源码

调用 ApplicationEventMulticaster接口 multicastEvent方法广播事件获取到 SimpleApplicationEventMulticaster就开始调用 multicastEvent方法来进行事件分发

public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {
  ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
    // 获取事件线程池，如果有则使用线程池异步执行
  Executor executor = getTaskExecutor();
    // getApplicationListeners 获取事件的所有监听器
  for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {
   if (executor != null) {
    executor.execute(() -> invokeListener(listener, event));
   }
   else {
        // 没有则直接执行
    invokeListener(listener, event);
   }
  }
 }

这里获取事件的所有监听器是通过方法getApplicationListeners来获取的，然后循环ApplicationListener去执行这里分同步执行和异步执行如果配置了事件执行线程池就使用线程池异步执行。否则就直接调用invokeListener方法

我们继续深入invokeListener方法看看是如何作广播的

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

这里也很简单，首先看有没有配置异常处理器，如果配置了就直接catch中执行作物处理器的方法，没有配置异常处理器就正常执行，所以我们还需要深入到doInvokeListener方法

doInvokeListener

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()) ||
     (event instanceof PayloadApplicationEvent &&
       matchesClassCastMessage(msg, ((PayloadApplicationEvent) event).getPayload().getClass()))) {
    // Possibly a lambda-defined listener which we could not resolve the generic event type for
    // -> let's suppress the exception.
    Log loggerToUse = this.lazyLogger;
    if (loggerToUse == null) {
     loggerToUse = LogFactory.getLog(getClass());
     this.lazyLogger = loggerToUse;
    }
    if (loggerToUse.isTraceEnabled()) {
     loggerToUse.trace("Non-matching event type for listener: " + listener, ex);
    }
   }
   else {
    throw ex;
   }
  }
 }

虽然这个方法很长，但是核心方法就是一行

   listener.onApplicationEvent(event);

这个方法是不是很熟悉，不是就我们实现ApplicationListener接口的方法吗

所以整个事件广播的流程就分析完了，整体源码还是比较简单的

监听器的获取

不难发现上面我们没有去分析getApplicationListeners获取事件的所有监听器整个方法我们没有去分析，因为整体流程和代码相对来说会复杂一些，所以我们单独抽出来分析

protected Collection<ApplicationListener<?>> getApplicationListeners(
   ApplicationEvent event, ResolvableType eventType) {
    // 获取事件源类 我们这里就是 ZouTest
  Object source = event.getSource();
  Class<?> sourceType = (source != null ? source.getClass() : null);
    // 构造一个事件缓存
  ListenerCacheKey cacheKey = new ListenerCacheKey(eventType, sourceType);

  // Potential new retriever to populate
  CachedListenerRetriever newRetriever = null;

  // Quick check for existing entry on ConcurrentHashMap  
    // 缓存相关玩意 开始为 null Map<ListenerCacheKey, CachedListenerRetriever>
  CachedListenerRetriever existingRetriever = this.retrieverCache.get(cacheKey);
  if (existingRetriever == null) {
   // Caching a new ListenerRetriever if possible
   if (this.beanClassLoader == null ||
     (ClassUtils.isCacheSafe(event.getClass(), this.beanClassLoader) &&
       (sourceType == null || ClassUtils.isCacheSafe(sourceType, this.beanClassLoader)))) {
        // 直接new一个 CachedListenerRetriever 然后丢到 retrieverCache 中
    newRetriever = new CachedListenerRetriever();
    existingRetriever = this.retrieverCache.putIfAbsent(cacheKey, newRetriever);
    if (existingRetriever != null) {
     newRetriever = null;  // no need to populate it in retrieveApplicationListeners
    }
   }
  }
    // 由于上面第一次没有caceh 所以不执行
  if (existingRetriever != null) {
   Collection<ApplicationListener<?>> result = existingRetriever.getApplicationListeners();
   if (result != null) {
    return result;
   }
   // If result is null, the existing retriever is not fully populated yet by another thread.
   // Proceed like caching wasn't possible for this current local attempt.
  }
    // 直接执行这里
  return retrieveApplicationListeners(eventType, sourceType, newRetriever);
 }

retrieveApplicationListeners

private Collection<ApplicationListener<?>> retrieveApplicationListeners(
   ResolvableType eventType, @Nullable Class<?> sourceType, @Nullable CachedListenerRetriever retriever) {

  List<ApplicationListener<?>> allListeners = new ArrayList<>();
  Set<ApplicationListener<?>> filteredListeners = (retriever != null ? new LinkedHashSet<>() : null);
  Set<String> filteredListenerBeans = (retriever != null ? new LinkedHashSet<>() : null);

  Set<ApplicationListener<?>> listeners;
  Set<String> listenerBeans;
  synchronized (this.defaultRetriever) {
   listeners = new LinkedHashSet<>(this.defaultRetriever.applicationListeners);
   listenerBeans = new LinkedHashSet<>(this.defaultRetriever.applicationListenerBeans);
  }

  // Add programmatically registered listeners, including ones coming
  // from ApplicationListenerDetector (singleton beans and inner beans).
  for (ApplicationListener<?> listener : listeners) {
   if (supportsEvent(listener, eventType, sourceType)) {
    if (retriever != null) {
     filteredListeners.add(listener);
    }
    allListeners.add(listener);
   }
  }

  // Add listeners by bean name, potentially overlapping with programmatically
  // registered listeners above - but here potentially with additional metadata.
  if (!listenerBeans.isEmpty()) {
   ConfigurableBeanFactory beanFactory = getBeanFactory();
   for (String listenerBeanName : listenerBeans) {
    try {
     if (supportsEvent(beanFactory, listenerBeanName, eventType)) {
      ApplicationListener<?> listener =
        beanFactory.getBean(listenerBeanName, ApplicationListener.class);
      if (!allListeners.contains(listener) && supportsEvent(listener, eventType, sourceType)) {
       if (retriever != null) {
        if (beanFactory.isSingleton(listenerBeanName)) {
         filteredListeners.add(listener);
        }
        else {
         filteredListenerBeans.add(listenerBeanName);
        }
       }
       allListeners.add(listener);
      }
     }
     else {
      // Remove non-matching listeners that originally came from
      // ApplicationListenerDetector, possibly ruled out by additional
      // BeanDefinition metadata (e.g. factory method generics) above.
      Object listener = beanFactory.getSingleton(listenerBeanName);
      if (retriever != null) {
       filteredListeners.remove(listener);
      }
      allListeners.remove(listener);
     }
    }
    catch (NoSuchBeanDefinitionException ex) {
     // Singleton listener instance (without backing bean definition) disappeared -
     // probably in the middle of the destruction phase
    }
   }
  }

  AnnotationAwareOrderComparator.sort(allListeners);
  if (retriever != null) {
   if (filteredListenerBeans.isEmpty()) {
    retriever.applicationListeners = new LinkedHashSet<>(allListeners);
    retriever.applicationListenerBeans = filteredListenerBeans;
   }
   else {
    retriever.applicationListeners = filteredListeners;
    retriever.applicationListenerBeans = filteredListenerBeans;
   }
  }
  return allListeners;
 }

这里获取到我们的事件监听器核心代码是在这里

可以看到DefaultListenerRetriever中的Set<ApplicationListener<?>> applicationListeners早就获取到了所有的事件监听器类

后面我们会分析如何注册事件的即applicationListeners中的值是如何填充的。这里我们先继续向下

继续在

for (ApplicationListener<?> listener : listeners) {
   if (supportsEvent(listener, eventType, sourceType)) {
    if (retriever != null) {
     filteredListeners.add(listener);
    }
    allListeners.add(listener);
   }
  }

这里就是获取所有监听事件com.zou.eventbus.ZouEvent的监听器

eventType: com.zou.eventbus.ZouEvent
sourceType: com.zou.springboot.ZouTest

至于泛型相关的判断处理主要是通过ResolvableType类处理的，这里就不展开细说了

这里我们可以看看allListeners中的值

里面多了个delegatingapplicationlistener不知道是干嘛用的，好像没啥用

  AnnotationAwareOrderComparator.sort(allListeners);

可以看到这里是支持注解排序的。这里就分析完获取事件的源码，还有很多其他细节没有分析到。但是太多了，限于篇幅就不继续展开。我们继续看看事件的注册

事件注册

上面我们知道我们获取事件是在这里

，所以我们直接查看哪里调用set方法就知道了

再更下去调用的地方有点多了

图文Debug深入研究下Spring Boot 事件发布监听源码所以我们直接debug，最终跟踪到注册的地方是在AbstractApplicationContext的finishBeanFactoryInitialization方法

注意有些博主说的是registerListeners();方法，实际上不是的，registerListeners();方法

图文Debug深入研究下Spring Boot 事件发布监听源码 registerListeners方法只是注册了Spring自己定义的一些事件,其次就是 applicationListenerBeans注册了我们自定义的Bean name

所以添加自定义事件最终是在finishBeanFactoryInitialization方法中。所以网上大部分源码分析其实是错误的，都是说在registerListeners中注册的

总结

总的来说Spring事件源码还是非常复杂的，我们这里只是分析了一下主线

事件的触发：事件源码比较简单，就是通过方法 getApplicationEventMulticaster获取时间分发器 ApplicationEventMulticaster然后通过 multicastEvent方法去分发事件，其中会检测是否配置了线程池，如果配置了线程池则异步去执行
事件监听器注册: 需要注意的是事件注册有两个方法 registerListeners是注册Spring自己默认提供的14个事件监听器。自定义事件监听器是在 registerListeners中注册的

vlambda博客
学习文章列表