Spring 事件发布

Posted 2020-11-29 蟹烟客

1、设计模式

基于观察者模式，主要方法为1 监听者注册 2 监听者注销 3 执行监听方法

 

2、使用篇

类结构图 

MsgEvent：事件对象

MsgListener：事件监听

MsgListener2：事件监听(使用注解方式实现)

MsgPublisher：事件发布器

SpringEventTest：单元测试类

测试代码

MsgEvent

@Data
public class MsgEvent extends ApplicationEvent{

    //消息ID
    private String msgId;
    //消息体
    private String payload;

    public MsgEvent(Object source) {
        super(source);
    }

}

MsgListener

@Component
public class MsgListener implements ApplicationListener<MsgEvent> {
    @Override
    public void onApplicationEvent(MsgEvent event) {
        System.out.println("listenter1 got message, ID:"+event.getMsgId()+", payload:"+event.getPayload());
    }
}

MsgListener2

@Component
public class MsgListener2 {

    @EventListener
    public void processMsg(MsgEvent event){
        System.out.println("listenter2 got message, ID:"+event.getMsgId()+", payload:"+event.getPayload());
    }
}

MsgPublisher

@Component
public class MsgPublisher implements ApplicationContextAware {
    //持有当前容器
    private ApplicationContext applicationContext;

    //模拟业务触发事件
    public void publish(){
        MsgEvent msgEvent = new MsgEvent("这是一条消息事件");
        msgEvent.setMsgId(UUID.randomUUID().toString());
        msgEvent.setPayload("事件消息体xxx");
        applicationContext.publishEvent(msgEvent);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

SpringEventTest

public class SpringEventTest extends BaseTest{

    @Autowired
    private MsgPublisher msgPublisher;


    @Test
    public void publish(){
        msgPublisher.publish();
    }
}

打印结果:

listenter2 got message, ID:5938c2c4-0cda-4abc-b187-9d347102e867, payload:事件消息体xxx
listenter1 got message, ID:5938c2c4-0cda-4abc-b187-9d347102e867, payload:事件消息体xxx

 

3、源码分析篇

流程简述：

1 注册Listener到容器中,集合存储 (本文忽略注册过程源码，着重发布事件和处理事件代码)

2 获取发布器SimpleApplicationEventMulticaster, 发布器在spring启动时会初始化 initApplicationEventMulticaster()方法 此处不细究

3 发布器根据事件source和事件类class从容器中获取监听器集合

4 遍历监听器集合, 并调用监听器EventListener的onApplication方法

 

类图展示：

 

 

以MsgPublisher为入口  跟踪方法 applicationContext.publishEvent(msgEvent)

@Component
public class MsgPublisher implements ApplicationContextAware {
    //持有当前容器
    private ApplicationContext applicationContext;

    //模拟业务触发事件
    public void publish(){
        MsgEvent msgEvent = new MsgEvent("这是source");
        msgEvent.setMsgId(UUID.randomUUID().toString());
        msgEvent.setPayload("事件消息体xxx");
        applicationContext.publishEvent(msgEvent);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

 

AbstractApplicationContext 持续追踪代码

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

    // Decorate event as an ApplicationEvent if necessary
    ApplicationEvent applicationEvent;
    //若继承自ApplicationEvent 则直接转Application类型
    if (event instanceof ApplicationEvent) {
        applicationEvent = (ApplicationEvent) event;
    }
    //非继承自ApplicationEvent 则标准化成 PayloadApplicationEvent
    //继承关系为 PayloadApplicationEvent<T> extends ApplicationEvent 所以最终还是ApplicationEvent方法
    else {
        applicationEvent = new PayloadApplicationEvent<Object>(this, event);
        if (eventType == null) {
            eventType = ((PayloadApplicationEvent)applicationEvent).getResolvableType();
        }
    }

    // 有可能事件广播器正在初始化 则存入事件列表延后处理
    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);
        }
    }
}


ApplicationEventMulticaster getApplicationEventMulticaster() throws IllegalStateException {
    if (this.applicationEventMulticaster == null) {
        throw new IllegalStateException("ApplicationEventMulticaster not initialized - " +
                "call \'refresh\' before multicasting events via the context: " + this);
    }
    //返回当前容器广播器 ApplicationEventMulticaster applicationEventMulticaster
    return this.applicationEventMulticaster;
}

SimpleApplicationEventMulticaster 发布器发布事件

//广播事件
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);
        }
    }
}

//筛选获取监听器
protected Collection<ApplicationListener<?>> getApplicationListeners(
            ApplicationEvent event, ResolvableType eventType) {

    //获取事件内容
    Object source = event.getSource();
    //获取事件source 可以理解为topic
    Class<?> sourceType = (source != null ? source.getClass() : null);
    //构建缓存KEY
    ListenerCacheKey cacheKey = new ListenerCacheKey(eventType, sourceType);

    //Map<ListenerCacheKey , ListenerRetriever> retrieverCache 
    //ListenerRetriever持有事件监听器集合 按ListenerCacheKey分组 
    ListenerRetriever retriever = this.retrieverCache.get(cacheKey);
    //若从缓存中获取到retriever 则直接返回持有的监听器集合
    if (retriever != null) {
        return retriever.getApplicationListeners();
    }

    if (this.beanClassLoader == null ||
            (ClassUtils.isCacheSafe(event.getClass(), this.beanClassLoader) &&
                    (sourceType == null || ClassUtils.isCacheSafe(sourceType, this.beanClassLoader)))) {
        // Fully synchronized building and caching of a ListenerRetriever
        synchronized (this.retrievalMutex) {
            retriever = this.retrieverCache.get(cacheKey);
            if (retriever != null) {
                return retriever.getApplicationListeners();
            }
            retriever = new ListenerRetriever(true);
            //根据eventType sourceType筛选监听器 并存入retriever中
            Collection<ApplicationListener<?>> listeners =
                    retrieveApplicationListeners(eventType, sourceType, retriever);
            //放入MAP缓存
            this.retrieverCache.put(cacheKey, retriever);
            return listeners;
        }
    }
    else {
        // No ListenerRetriever caching -> no synchronization necessary
        return retrieveApplicationListeners(eventType, sourceType, null);
    }
}
//调用监听器 事件处理
protected void invokeListener(ApplicationListener listener, ApplicationEvent event) {
    ErrorHandler errorHandler = getErrorHandler();
    if (errorHandler != null) {
        try {
            //执行监听器的onApplicationEvent方法
            listener.onApplicationEvent(event);
        }
        catch (Throwable err) {
            errorHandler.handleError(err);
        }
    }
    else {
        try {
            //执行监听器的onApplicationEvent方法
            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;
            }
        }
    }
}