Spring事件機(jī)制問題排查

alphahans 發(fā)布于2019-08-16 16:11 / 3085人閱讀

摘要：前言之前使用的事件機(jī)制來改造系統(tǒng)，完成了部分模塊的解耦。但是實際使用時卻發(fā)現(xiàn)存在以下問題當(dāng)批量推送時，如果在處理的過程中拋出異常，則會導(dǎo)致后續(xù)的推送中斷。但是實際上卻是拋出異常會導(dǎo)致后續(xù)事件的推送中斷。

前言

之前使用Spring的事件機(jī)制來改造系統(tǒng)，完成了部分模塊的解耦。但是實際使用時卻發(fā)現(xiàn)存在以下問題：

當(dāng)ApplicationEventPublisher批量推送ApplicationEvent時，如果ApplicationListener在處理的過程中拋出異常，則會導(dǎo)致后續(xù)的推送中斷。

PS：Spring版本為5.1.5.RELEASE

下面將會展示一個復(fù)盤的示例

復(fù)盤示例 自定義事件

import org.springframework.context.ApplicationEvent;

/**
 * 自定義事件
 * @author RJH
 * create at 2018/10/29
 */
public class SimpleEvent extends ApplicationEvent {

    private int i;
    /**
     * Create a new ApplicationEvent.
     *
     * @param source the object on which the event initially occurred (never {@code null})
     */
    public SimpleEvent(Object source) {
        super(source);
        i=Integer.valueOf(source.toString());
    }

    public int getI() {
        return i;
    }

}

事件監(jiān)聽器

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

/**
 * 自定義事件監(jiān)聽器
 * @author RJH
 * create at 2018/10/29
 */
@Component
public class SimpleEventListener implements ApplicationListener {

    @Override
    public void onApplicationEvent(SimpleEvent event) {
        if(event.getI()%10==0){
            throw new RuntimeException();
        }
        System.out.println("Time:"+event.getTimestamp()+" event:"+event.getSource());
    }

}

事件推送

import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

/**
 * 事件推送
 * @author RJH
 * create at 2018/10/29
 */
public class EventApplication {

    public static void main(String[] args) {
        //掃描特定package
        ApplicationContext context=new AnnotationConfigApplicationContext("com.rjh.event");
        for(int i=1;i<=100;i++){//批量推送事件
            context.publishEvent(new SimpleEvent(i));
        }
    }
}

運(yùn)行結(jié)果

Time:1553607971143 event:1
Time:1553607971145 event:2
Time:1553607971145 event:3
Time:1553607971145 event:4
Time:1553607971145 event:5
Time:1553607971145 event:6
Time:1553607971146 event:7
Time:1553607971146 event:8
Time:1553607971146 event:9
Exception in thread "main" java.lang.RuntimeException
    at com.rjh.event.SimpleEventListener.onApplicationEvent(SimpleEventListener.java:17)
    at com.rjh.event.SimpleEventListener.onApplicationEvent(SimpleEventListener.java:11)
    at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:172)
    at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:165)
    at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:139)
    at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:393)
    at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:347)
    at com.rjh.event.EventApplication.main(EventApplication.java:17)

分析

期待結(jié)果為SimpleEventListener拋出異常不影響EventApplication中后續(xù)事件的推送。但是實際上卻是SimpleEventListener拋出異常會導(dǎo)致EventApplication后續(xù)事件的推送中斷。從這里可以看出事件的推送和事件的監(jiān)聽是同步阻塞進(jìn)行，而并非是異步。詳細(xì)可以參考文檔中的介紹：

Notice that ApplicationListener is generically parameterized with the type of your custom event (BlackListEvent in the preceding example). This means that the onApplicationEvent() method can remain type-safe, avoiding any need for downcasting. You can register as many event listeners as you wish, but note that, by default, event listeners receive events synchronously. This means that the publishEvent() method blocks until all listeners have finished processing the event. One advantage of this synchronous and single-threaded approach is that, when a listener receives an event, it operates inside the transaction context of the publisher if a transaction context is available. If another strategy for event publication becomes necessary, See the javadoc for Spring’s ApplicationEventMulticaster interface.

解決辦法

將事件監(jiān)聽改造為異步處理，這里將會展示基于JavaConfig即注解的解決方案

開啟異步

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

/**
 * 開啟異步服務(wù)配置類
 * @author RJH
 * create at 2019-03-26
 */
@EnableAsync
@Configuration
public class AsyncConfig {

}

異步事件監(jiān)聽

import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * 異步事件監(jiān)聽
 * @author RJH
 * create at 2019-03-26
 */
@Component
public class AsyncSimpleEventListener {

    @EventListener
    @Async
    public void handleEvent(SimpleEvent event){
        if(event.getI()%10==0){
            throw new RuntimeException();
        }
        System.out.println("Time:"+event.getTimestamp()+" event:"+event.getSource());
    }
}

運(yùn)行結(jié)果

Time:1553614469990 event:1
Time:1553614470007 event:72
Time:1553614470006 event:64
Time:1553614470006 event:67
Time:1553614470007 event:73
Time:1553614470007 event:71
Time:1553614470007 event:75
Time:1553614470006 event:68
Time:1553614470007 event:69
Time:1553614470006 event:62
Time:1553614470005 event:61
Time:1553614470006 event:63
Time:1553614470006 event:65
Time:1553614470007 event:74
Time:1553614470006 event:66
Time:1553614470005 event:59
Time:1553614470005 event:57
Time:1553614470005 event:55
Time:1553614470005 event:58
Time:1553614470004 event:51
Time:1553614470004 event:52
Time:1553614470002 event:43
Time:1553614470004 event:53
Time:1553614470002 event:38
Time:1553614470001 event:36
Time:1553614470004 event:54
Time:1553614470001 event:33
Time:1553614470000 event:29
Time:1553614470000 event:27
Time:1553614470005 event:56
Time:1553614469999 event:23
Time:1553614469999 event:22
Time:1553614469999 event:21
三月 26, 2019 11:34:30 下午 org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler handleUncaughtException
嚴(yán)重: Unexpected error occurred invoking async method: public void com.rjh.event.AsyncSimpleEventListener.handleEvent(com.rjh.event.SimpleEvent)
Time:1553614470000 event:24java.lang.RuntimeException
    at com.rjh.event.AsyncSimpleEventListener.handleEvent(AsyncSimpleEventListener.java:19)

    at com.rjh.event.AsyncSimpleEventListener$$FastClassBySpringCGLIB$$61742dbf.invoke()
Time:1553614469998 event:15    at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:736)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)

    at org.springframework.aop.interceptor.AsyncExecutionInterceptor$1.call(AsyncExecutionInterceptor.java:115)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
...內(nèi)容過長省略部分結(jié)果

分析

改造為異步執(zhí)行后，事件監(jiān)聽就由線程池進(jìn)行處理，此處還可以通過自定義線程池，并設(shè)置異常處理器來處理未捕獲的異常。

參考資料

https://docs.spring.io/spring...