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

2024-02-10 13:11:10

简介
在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完成事件发布。

源码展示（https://www.xiaoyuani.com/）
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;
	}
}

Spring 事件监听机制及原理分析
从上面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 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 容器初始化前后都可能进行广播事件

上一篇：jmeter（十三）常见问题及解决方法


下一篇：[从源码学设计]蚂蚁金服SOFARegistry之消息总线