觀察者模式在spring中的應(yīng)用
1.1 定義
指多個(gè)對(duì)象間存在一對(duì)多的依賴(lài)關(guān)系，當(dāng)一個(gè)對(duì)象的狀態(tài)發(fā)生改變時(shí)，所有依賴(lài)于它的對(duì)象都得到通知并被自動(dòng)更新。這種模式有時(shí)又稱(chēng)作發(fā)布-訂閱模式、模型-視圖模式，它是對(duì)象行為型模式。
1.2 角色介紹
在觀察者模式中，有以下幾個(gè)角色。
主題也叫被觀察者（Subject）：
定義被觀察者必須實(shí)現(xiàn)的職責(zé)，
它能動(dòng)態(tài)的增加取消觀察者，它一般是抽象類(lèi)或者是實(shí)現(xiàn)類(lèi)，僅僅完成作為被觀察者必須實(shí)現(xiàn)的職責(zé)：
管理觀察者并通知觀察者。
觀察者（Observer）：
觀察者接收到消息后，即進(jìn)行更新操作，對(duì)接收到的信息進(jìn)行處理。
具體的被觀察者（ConcreteSubject）：
定義被觀察者自己的業(yè)務(wù)邏輯，同時(shí)定義對(duì)哪些事件進(jìn)行通知。
具體的觀察者（ConcreteObserver）：
具體的觀察者，每個(gè)觀察者接收到消息后的處理反應(yīng)是不同的，每個(gè)觀察者都有自己的處理邏輯。
1.3 觀察者模式的適用場(chǎng)景
對(duì)象間存在一對(duì)多關(guān)系，一個(gè)對(duì)象的狀態(tài)發(fā)生改變會(huì)影響其他對(duì)象。
當(dāng)一個(gè)抽象模型有兩個(gè)方面，其中一個(gè)方面依賴(lài)于另一方面時(shí)，可將這二者封裝在獨(dú)立的對(duì)象中以使它們可以各自獨(dú)立地改變和復(fù)用。
實(shí)現(xiàn)類(lèi)似廣播機(jī)制的功能，不需要知道具體收聽(tīng)者，只需分發(fā)廣播，系統(tǒng)中感興趣的對(duì)象會(huì)自動(dòng)接收該廣播。
多層級(jí)嵌套使用，形成一種鏈?zhǔn)接|發(fā)機(jī)制，使得事件具備跨域（跨越兩種觀察者類(lèi)型）通知。
2.1 spring的事件監(jiān)聽(tīng)機(jī)制
Spring事件機(jī)制是觀察者模式的實(shí)現(xiàn)。ApplicationContext中事件處理是由ApplicationEvent類(lèi)和ApplicationListener接口來(lái)提供的。如果一個(gè)Bean實(shí)現(xiàn)了ApplicationListener接口，并且已經(jīng)發(fā)布到容器中去，每次ApplicationContext發(fā)布一個(gè)ApplicationEvent事件，這個(gè)Bean就會(huì)接到通知。ApplicationEvent 事件的發(fā)布需要顯示觸發(fā)，要么 Spring 觸發(fā)，要么我們編碼觸發(fā)。spring內(nèi)置事件由spring觸發(fā)。我們先來(lái)看一下，如何自定義spring事件，并使其被監(jiān)聽(tīng)和發(fā)布。
2.1.1 事件
事件，ApplicationEvent，該抽象類(lèi)繼承了EventObject，EventObject是JDK中的類(lèi)，并建議所有的事件都應(yīng)該繼承自EventObject。
public abstract class ApplicationEvent extends EventObject {private static final long serialVersionUID = 7099057708183571937L;private final long timestamp = System.currentTimeMillis();public ApplicationEvent(Object source) {super(source);}public final long getTimestamp() {return this.timestamp;}}
2.1.2 監(jiān)聽(tīng)器
ApplicationListener，是一個(gè)接口，該接口繼承了EventListener接口。EventListener接口是JDK中的，建議所有的事件監(jiān)聽(tīng)器都應(yīng)該繼承EventListener。
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {void onApplicationEvent(E var1);}
2.1.3 事件發(fā)布器
ApplicationEventPublisher，ApplicationContext繼承了該接口，在ApplicationContext的抽象實(shí)現(xiàn)類(lèi)AbstractApplicationContext中做了實(shí)現(xiàn)下面我們來(lái)看一下
org.springframework.context.support.AbstractApplicationContext#publishEvent(java.lang.Object, org.springframework.core.ResolvableType)
protected void publishEvent(Object event, @Nullable ResolvableType eventType) {Assert.notNull(event, "Event must not be null");ApplicationEvent applicationEvent;if (event instanceof ApplicationEvent) {applicationEvent = (ApplicationEvent) event;}else {applicationEvent = new PayloadApplicationEvent<>(this, event);if (eventType == null) {eventType = ((PayloadApplicationEvent<?>) applicationEvent).getResolvableType();}}if (this.earlyApplicationEvents != null) {this.earlyApplicationEvents.add(applicationEvent);}else {//獲取當(dāng)前注入的發(fā)布器,執(zhí)行發(fā)布方法getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);}if (this.parent != null) {if (this.parent instanceof AbstractApplicationContext) {((AbstractApplicationContext) this.parent).publishEvent(event, eventType);}else {this.parent.publishEvent(event);}}}
我們可以看到，AbstractApplicationContext中publishEvent方法最終執(zhí)行發(fā)布事件的是ApplicationEventMulticaster#multicastEvent方法，下面我們?cè)賮?lái)一起看一下multicastEvent方法:
public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));Executor executor = getTaskExecutor();//拿到所有的監(jiān)聽(tīng)器,如果異步執(zhí)行器不為空,異步執(zhí)行for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {if (executor != null) {executor.execute(() -> invokeListener(listener, event));}else {//執(zhí)行監(jiān)聽(tīng)方法invokeListener(listener, event);}}}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);}}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())) {Log logger = LogFactory.getLog(getClass());if (logger.isTraceEnabled()) {logger.trace("Non-matching event type for listener: " + listener, ex);}}else {throw ex;}}}
上面介紹了非spring內(nèi)置的事件發(fā)布和監(jiān)聽(tīng)執(zhí)行流程。總結(jié)一下
定義一個(gè)事件,該事件繼承ApplicationEvent
定義一個(gè)監(jiān)聽(tīng)器,實(shí)現(xiàn)ApplicationListener接口
定義一個(gè)事件發(fā)布器,實(shí)現(xiàn)ApplicationEventPublisherAware接口
調(diào)用ApplicationEventPublisher#publishEvent方法
2.2 spring事件實(shí)現(xiàn)原理
上面我們講到了spring事件的發(fā)布，那么spring事件發(fā)布之后，spring是如何根據(jù)事件找到事件對(duì)應(yīng)的監(jiān)聽(tīng)器呢？我們一起來(lái)探究一下。
spring的容器初始化過(guò)程想必大家都已十分了解，這里就不過(guò)多贅述，我們直接看refresh方法在refresh方法中，有這樣兩個(gè)方法，initApplicationEventMulticaster()和registerListeners()
我們先來(lái)看initApplicationEventMulticaster()方法
2.2.1 initApplicationEventMulticaster()
org.springframework.context.support.AbstractApplicationContext#initApplicationEventMulticaster
public static final String APPLICATION_EVENT_MULTICASTER_BEAN_NAME = "applicationEventMulticaster";protected void initApplicationEventMulticaster() {//獲得beanFactoryConfigurableListableBeanFactory beanFactory = getBeanFactory();//BeanFactory中是否有ApplicationEventMulticasterif (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 {//如果BeanFactory中不存在，就創(chuàng)建一個(gè)SimpleApplicationEventMulticasterthis.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() + "]");}}}
上述代碼我們可以看出，spring先從BeanFactory中獲取applicationEventMulticaster如果為空，則直接創(chuàng)建SimpleApplicationEventMulticaster
2.2.2 registerListeners()
org.springframework.context.support.AbstractApplicationContext#registerListeners
registerListeners 是將各種實(shí)現(xiàn)了 ApplicationListener 的監(jiān)聽(tīng)器注冊(cè)到 ApplicationEventMulticaster 事件廣播器中
protected void registerListeners() {for (ApplicationListener<?> listener : getApplicationListeners()) {getApplicationEventMulticaster().addApplicationListener(listener);}String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false);for (String listenerBeanName : listenerBeanNames) {//把監(jiān)聽(tīng)器注冊(cè)到事件發(fā)布器上getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName);}Set<ApplicationEvent> earlyEventsToProcess = this.earlyApplicationEvents;this.earlyApplicationEvents = null;if (!CollectionUtils.isEmpty(earlyEventsToProcess)) {//如果內(nèi)置監(jiān)聽(tīng)事件集合不為空for (ApplicationEvent earlyEvent : earlyEventsToProcess) {//執(zhí)行spring內(nèi)置的監(jiān)聽(tīng)方法getApplicationEventMulticaster().multicastEvent(earlyEvent);}}}
這里解釋一下earlyApplicationListeners
earlyApplicationListeners的本質(zhì)還是ApplicationListener。Spring單例Ban的實(shí)例化是在Refresh階段實(shí)例化的，那么用戶(hù)自定義的一些ApplicationListener組件自然也是在這個(gè)階段才初始化，但是Spring容器啟動(dòng)過(guò)程中，在Refresh完成之前還有很多事件：如Spring上下文環(huán)境準(zhǔn)備等事件，這些事件又是Spring容器啟動(dòng)必須要監(jiān)聽(tīng)的。所以Spring定義了一個(gè)earlyApplicationListeners集合，這個(gè)集合中的Listener在factories文件中定義好，在容器Refresh之前預(yù)先實(shí)例化好，然后就可以監(jiān)聽(tīng)Spring容器啟動(dòng)過(guò)程中的所有事件。
當(dāng)registerListeners方法執(zhí)行完成,我們的監(jiān)聽(tīng)器已經(jīng)添加到多播器SimpleApplicationEventMulticaster中了，并且earlyEvent 早期事件也已經(jīng)執(zhí)行完畢。但是我們發(fā)現(xiàn),如果自定義了一個(gè)監(jiān)聽(tīng)器去監(jiān)聽(tīng)spring內(nèi)置的事件,此時(shí)并沒(méi)有被執(zhí)行，那我們注冊(cè)的監(jiān)聽(tīng)器是如何被執(zhí)行的呢？答案在finishRefresh方法中。
2.2.3 finishRefresh
org.springframework.context.support.AbstractApplicationContext#finishRefresh
protected void finishRefresh() {clearResourceCaches();initLifecycleProcessor();getLifecycleProcessor().onRefresh();//容器中的類(lèi)全部初始化完畢后，觸發(fā)刷新事件publishEvent(new ContextRefreshedEvent(this));LiveBeansView.registerApplicationContext(this);}
如果我們想要實(shí)現(xiàn)在spring容器中所有bean創(chuàng)建完成后做一些擴(kuò)展功能，我們就可以實(shí)現(xiàn)ApplicationListener
2.2.4 spring中異步的監(jiān)聽(tīng)執(zhí)行機(jī)制
我們回過(guò)頭來(lái)看一下ApplicationEventMulticaster#pushEvent方法
protected void publishEvent(Object event, @Nullable ResolvableType eventType) {Assert.notNull(event, "Event must not be null");ApplicationEvent applicationEvent;if (event instanceof ApplicationEvent) {applicationEvent = (ApplicationEvent) event;}else {applicationEvent = new PayloadApplicationEvent<>(this, event);if (eventType == null) {eventType = ((PayloadApplicationEvent<?>) applicationEvent).getResolvableType();}}if (this.earlyApplicationEvents != null) {this.earlyApplicationEvents.add(applicationEvent);}else {//獲取當(dāng)前注入的發(fā)布器,執(zhí)行發(fā)布方法getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);}if (this.parent != null) {if (this.parent instanceof AbstractApplicationContext) {((AbstractApplicationContext) this.parent).publishEvent(event, eventType);}else {this.parent.publishEvent(event);}}
最終執(zhí)行發(fā)布事件的是ApplicationEventMulticaster#multicastEvent方法，下面我們?cè)賮?lái)一起看一下multicastEvent方法
public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));Executor executor = getTaskExecutor();//拿到所有的監(jiān)聽(tīng)器,如果異步執(zhí)行器不為空,異步執(zhí)行for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {if (executor != null) {executor.execute(() -> invokeListener(listener, event));}else {//執(zhí)行監(jiān)聽(tīng)方法invokeListener(listener, event);}}}
可以看到，異步事件通知主要依靠SimpleApplicationEventMulticaster 類(lèi)中的Executor去實(shí)現(xiàn)的，如果這個(gè)變量不配置的話(huà)默認(rèn)事件通知是同步的， 否則就是異步通知了，要實(shí)現(xiàn)同時(shí)支持同步通知和異步通知就得從這里下手；我們上文已經(jīng)分析過(guò)了在initApplicationEventMulticaster方法中有這樣一段代碼
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 + "]");}}
如果BeanFactory中已經(jīng)有了SimpleApplicationEventMulticaster則不會(huì)重新創(chuàng)建，那么我們可以在spring中注冊(cè)一個(gè)SimpleApplicationEventMulticaster并且向其中注入對(duì)應(yīng)的Executor這樣我們就可以得到一個(gè)異步執(zhí)行監(jiān)聽(tīng)的SimpleApplicationEventMulticaster了，我們的通知就會(huì)通過(guò)Executor異步執(zhí)行。這樣可以大大提高事件發(fā)布的效率。
在springboot項(xiàng)目中我們可以增加一個(gè)配置類(lèi)來(lái)實(shí)現(xiàn)
public class Config {(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME)public SimpleApplicationEventMulticaster simpleApplicationEventMulticaster(){SimpleApplicationEventMulticaster simpleApplicationEventMulticaster = new SimpleApplicationEventMulticaster();simpleApplicationEventMulticaster.setTaskExecutor(taskExecutor());return simpleApplicationEventMulticaster;}public TaskExecutor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(20);executor.setQueueCapacity(100);executor.setKeepAliveSeconds(300);executor.setThreadNamePrefix("thread-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);return executor;}}
spring項(xiàng)目中我們也可以增加如下xml配置
<!--定義事件異步處理--><bean id="commonTaskExecutor"class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"><!-- 線程池維持處于Keep-alive狀態(tài)的線程數(shù)量。如果設(shè)置了allowCoreThreadTimeOut為true，該值可能為0。并發(fā)線程數(shù),想達(dá)到真正的并發(fā)效果,最好對(duì)應(yīng)CPU的線程數(shù)及核心數(shù) --><property name="corePoolSize" value="5" /><!-- 最大線程池容量 --><property name="maxPoolSize" value="20" /><!-- 超過(guò)最大線程池容量后,允許的線程隊(duì)列數(shù) --><property name="queueCapacity" value="100" /><!-- 線程池維護(hù)線程所允許的空閑時(shí)間 .單位毫秒，默認(rèn)為60s，超過(guò)這個(gè)時(shí)間后會(huì)將大于corePoolSize的線程關(guān)閉，保持corePoolSize的個(gè)數(shù) --><property name="keepAliveSeconds" value="300" /><!-- 允許核心線程超時(shí): false（默認(rèn)值）不允許超時(shí)，哪怕空閑；true則使用keepAliveSeconds來(lái)控制等待超時(shí)時(shí)間，最終corePoolSize的個(gè)數(shù)可能為0 --><property name="allowCoreThreadTimeOut" value="true" /><!-- 線程池對(duì)拒絕任務(wù)(無(wú)線程可用)的處理策略 --><property name="rejectedExecutionHandler"><bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /><!-- java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy:主線程直接執(zhí)行該任務(wù)，執(zhí)行完之后嘗試添加下一個(gè)任務(wù)到線程池中 --><!-- java.util.concurrent.ThreadPoolExecutor$AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 --></property></bean><!--名字必須是applicationEventMulticaster，因?yàn)锳bstractApplicationContext默認(rèn)找個(gè)--><bean id="applicationEventMulticaster" class="org.springframework.context.event.SimpleApplicationEventMulticaster"><!--注入任務(wù)執(zhí)行器 這樣就實(shí)現(xiàn)了異步調(diào)用--><property name="taskExecutor" ref="commonTaskExecutor"></property></bean>
本文主要講解了觀察者模式在spring中的應(yīng)用及事件監(jiān)聽(tīng)機(jī)制，JDK 也有實(shí)現(xiàn)提供事件監(jiān)聽(tīng)機(jī)制Spring 的事件機(jī)制也是基于JDK 來(lái)擴(kuò)展的。Spring 的事件機(jī)制默認(rèn)是同步阻塞的，想要提升對(duì)應(yīng)的效率要考慮異步事件。
-end-