Spring的监听器和多播器

一. 简单了解

• 监听器：观察者角色，用来监听对应的事件，做相应的处理
• 事件：被观察者角色，监听器重点监听对象
• 事件源：用来创建并发布事件
• 多播器（广播器）：拥有监听器对象，提供发布事件的功能，遍历监听器，监听器来处理自己需要处理的事件

二. 自定义Spring的监听事件

2.1 Spring容器刷新完成事件

这个例子是在Spring容器刷新完成后会调用该事件

import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

@Component
public class MyListener implements ApplicationListener<ContextRefreshedEvent> {


    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        System.out.println("监听到：" + event.toString());
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.2 自定义事件

下面这个例子是我们自定义的事件类型，需要我们手动触发事件

import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ApplicationContextEvent;

/**
 * @author jingchuan
 */
public abstract class MyEvent extends ApplicationContextEvent {

    /**
     * 事件描述
     */
    private String description;

    public MyEvent(ApplicationContext source, String description) {
        super(source);
        this.description = description;
    }

    public String getDescription() {
        return description;
    }

    public void setDescription(String description) {
        this.description = description;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import org.springframework.context.ApplicationContext;

/**
 * 自定义事件
 * @author jingchuan
 */
public class HungryEvent extends MyEvent {

    public HungryEvent(ApplicationContext source, String description) {
        super(source, description);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13

import org.springframework.context.ApplicationContext;

/**
 * 自定义事件
 * @author jingchuan
 */
public class ThirstyEvent extends MyEvent {

    public ThirstyEvent(ApplicationContext source, String description) {
        super(source, description);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

/**
 * 自定义监听器，对事件做出相应
 * 泛型为此监听器的监听类型
 * @author jingchuan
 */
@Component
public class MyListener implements ApplicationListener<MyEvent> {

    @Override
    public void onApplicationEvent(MyEvent event) {
        System.out.println("监听到：" + event.getDescription());
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

启动类启动完成后触发事件：

import com.ygz.test1.event.HungryEvent;
import com.ygz.test1.event.ThirstyEvent;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class Test1Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext applicationContext = SpringApplication.run(Test1Application.class, args);
        applicationContext.publishEvent(new HungryEvent(applicationContext,"hungry"));
        applicationContext.publishEvent(new ThirstyEvent(applicationContext,"thirsty"));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

三. 源码分析

ApplicationEvent 是对 Java EventObject 的扩展，表示 Spring 中的事件，Spring 中的所有事件都要基于其进行扩展。
ApplicationListener 是 Spring 事件的监听器，可以用来接收上述中说明的 ApplicationEvent 事件，所有的监听器都必须实现该接口。
ApplicationContextEvent：Spring 应用上下文生命周期中发布的事件，对应不同的生命周期，其事件类型如下。

• ContextStartedEvent：应用上下文启动
• ContextStoppedEvent：应用上下文停止
• ContextRefreshedEvent：应用上下文刷新
• ContextClosedEvent：应用上下文关闭

事件发布器用于用于发布 ApplicationEvent 事件，发布后 ApplicationListener 才能监听到事件进行处理。Spring 的事件发布器包括 ApplicationEventPublisher 和 ApplicationEventMulticaster 两种。

ApplicationEventMulticaster是spring中事件广播器接口，负责事件的广播发布。ApplicationEventMulticaster 是 ApplicationEventPublisher 的底层实现，ApplicationEventMulticaster 的获取除了可以通过依赖注入，还可以通过依赖查找的方式。Spring默认使用SimpleApplicationEventMulticaster多播器

Spring对监听器和多播器的处理都在org.springframework.context.support.AbstractApplicationContext#refresh方法中：

1. initApplicationEventMulticaster(); 初始化多播器
2. registerListeners(); 注册事件到多播器
3. finishRefresh(); 中的 publishEvent(new ContextRefreshedEvent(this)); 触发事件

接下来依次分析

3.1 初始化多播器

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 {
		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() + "]");
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

这段代码内容很简单，在容器中查找applicationEventMulticaster，没有的话创建一个SimpleApplicationEventMulticaster，所以默认的多播器就是SimpleApplicationEventMulticaster

3.2 注册事件到多播器

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!
	// 在容器中获取所有实现了ApplicationListener接口的bd，并将beaName加入到多播器中
	// 这些都还没有实例化 实际是添加到了org.springframework.context.event.AbstractApplicationEventMulticaster.DefaultListenerRetriever#applicationListenerBeans中
	String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false);
	for (String listenerBeanName : listenerBeanNames) {
		getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName);
	}

	// Publish early application events now that we finally have a multicaster...
	// 发布早期的监听事件集合
    // 在refresh() 第一个方法 prepareRefresh() 中初始化了
	Set<ApplicationEvent> earlyEventsToProcess = this.earlyApplicationEvents;
	this.earlyApplicationEvents = null;
	if (!CollectionUtils.isEmpty(earlyEventsToProcess)) {
		for (ApplicationEvent earlyEvent : earlyEventsToProcess) {
			getApplicationEventMulticaster().multicastEvent(earlyEvent);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

3.3 容器刷新完成，发送ContextRefreshedEvent事件

protected void finishRefresh() {
	// Clear context-level resource caches (such as ASM metadata from scanning).
	clearResourceCaches();

	// Initialize lifecycle processor for this context.
	initLifecycleProcessor();

	// Propagate refresh to lifecycle processor first.
	getLifecycleProcessor().onRefresh();

	// Publish the final event.
	// 出发容器刷新完成事件
	publishEvent(new ContextRefreshedEvent(this));

	// Participate in LiveBeansView MBean, if active.
	LiveBeansView.registerApplicationContext(this);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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) {
		// 把ContextRefreshedEvent变成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 {
		// 将给定的ContextRefreshedEvent事件多播到适当的侦听器
		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);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

跟进multicastEvent(applicationEvent, eventType);

public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {
	// 入参的event为ContextRefreshedEvent  	eventType为null
	// 所以先确定eventType的值 因为eventType为null  所以获取的type为ContextRefreshedEvent的ResolvableType 
	ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
	// 这里是异步监听的处理  获取执行器的   默认是同步监听处理的
	Executor executor = getTaskExecutor();
	// getApplicationListeners(event, type) 是获取所有符合事件类型的监听  然后循环依次通知
	for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {
		if (executor != null) {
			executor.execute(() -> invokeListener(listener, event));
		}
		else {
			// 如果不是异步  那么在这里执行监听通知
			// 实际就是执行ApplicationListener的onApplicationEvent方法   也就是我们监听器的自定义监听内容
			invokeListener(listener, event);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

跟进getApplicationListeners(event, type)来详细看：

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

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

	// Quick check for existing entry on ConcurrentHashMap
	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)))) {
			newRetriever = new CachedListenerRetriever();
			existingRetriever = this.retrieverCache.putIfAbsent(cacheKey, newRetriever);
			if (existingRetriever != null) {
				newRetriever = null;  // no need to populate it in retrieveApplicationListeners
			}
		}
	}

	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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

上面这个方法主要就是先做一遍筛选，返回与给定事件类型匹配的 ApplicationListener 的集合。不匹配的会提前被排除在外，实际还是要看：retrieveApplicationListeners(eventType, sourceType, newRetriever);

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).
	// 循环所有已有的监听  通过supportsEvent(listener, eventType, sourceType) 筛选出匹配的监听器 
	// 把筛选出来的加入到allListeners 最后返回
	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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

经过筛选获取到的监听，就可以在org.springframework.context.event.SimpleApplicationEventMulticaster#multicastEvent(org.springframework.context.ApplicationEvent, org.springframework.core.ResolvableType)中进行循环触发。

在上面的例子中，如果是容器刷新的事件，就会被自动触发，因为在publishEvent(new ContextRefreshedEvent(this));给定了触发的就是ContextRefreshedEvent事件。
但是如果是自定义的事件类型，那么就要在容器刷新完成后手动publishEvent进行事件发送。