简介
在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完成事件发布。
源码展示
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;
}
}
从上面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 <E> 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 容器初始化前后都可能进行广播事件