看我如何教我同事處理SpringEvent接口超時(shí)問題
最近線上的一個(gè)接口前端訪問竟然超時(shí)了，前端設(shè)置的超時(shí)時(shí)間是10s,也就是說這個(gè)接口在10s內(nèi)都還沒有返回?cái)?shù)據(jù)，同事去看了說這個(gè)接口就是一個(gè)很簡(jiǎn)單的新增數(shù)據(jù)接口，不應(yīng)該導(dǎo)致超時(shí)的呀？
我看了下這個(gè)接口的邏輯,在結(jié)合日志初步定位是因?yàn)镾pring Event中調(diào)用其他服務(wù)接口導(dǎo)致的超時(shí)。
問題追蹤
首先看下這個(gè)接口調(diào)用的主要邏輯
public void saveData(MergeDataParam param) {
MergeData data = convert(param);
dataMapper.insert(data);
applicationContext.publishEvent(new MergeDataEvent(new MergeDataMessage(param.getDataId())));
}
@EventListener
public void notifyOnboardServiceWhenMergeData(MergeDataEvent event) {
MergeDataMessage msg = (MergeDataMessage)event.getSource();
//記錄日志
log.info("notify onbaord service when merge data, data id is {}", msg.getDataId());
NotifyData notifyData = getData(msg.getDataId());
Response<Result> response = remoteOnboardService.notify(notifyData);
// 省略其他業(yè)務(wù)代碼
log.info("notify result is {}", JSON.toJSONString(response));
}確實(shí)邏輯比較簡(jiǎn)單，主要就是保存數(shù)據(jù)，然后通知下游系統(tǒng)，但是從日志上來看，日志只打印了 notify onbaord service when merge data, data id is 這里的內(nèi)容，然后前端就超時(shí)了。
接著我又去看了下游服務(wù)的實(shí)現(xiàn)，發(fā)現(xiàn)確實(shí)有一些超時(shí)業(yè)務(wù)邏輯，但是下游服務(wù)是別人寫的，我們改不了。
同事還一直以為這個(gè)事件是異步的,我告訴他不是的，如果要想變成異步的需要自己配置。
要想配置成異步的有兩種方式,一種是通過@Async注解,另一種方式是在EventMulticaster中指定線程池。
@Async注解使用
首先我們需要開啟異步支持，然后指定一個(gè)異步執(zhí)行器，不然使用的線程池配置是默認(rèn)的
@Configuration
@EnableAsync
public class AsyncConfig {
@Bean(name = "asyncExecutor")
public Executor asyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.setMaxPoolSize(5);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("NotifyAsyncThread-");
executor.initialize();
return executor;
}
}然后在我們事件監(jiān)聽器上加上@Async注解
@Async("asyncExecutor")
@EventListener
public void notifyOnboardServiceWhenMergeData(MergeDataEvent event) {
MergeDataMessage msg = (MergeDataMessage)event.getSource();
//記錄日志
log.info("notify onbaord service when merge data, data id is {}", msg.getDataId());
NotifyData notifyData = getData(msg.getDataId());
Response<Result> response = remoteOnboardService.notify(notifyData);
// 省略其他業(yè)務(wù)代碼
log.info("notify result is {}", JSON.toJSONString(response));
}然后其他代碼不用改變，此時(shí)我們的監(jiān)聽器就可以異步執(zhí)行了。需要記住使用這個(gè)注解有一些注意事項(xiàng)
- 異步方法不能返回值。如果需要返回值，可以考慮使用 Future<T> 或 CompletableFuture<T>。
- 異步方法中拋出的異常不會(huì)被調(diào)用者直接捕獲。確保適當(dāng)處理異常，或考慮使用 AsyncUncaughtExceptionHandler。
- 如果在同一個(gè)類中調(diào)用異步方法，直接調(diào)用不會(huì)觸發(fā)異步行為。這是因?yàn)镾pring的AOP代理機(jī)制導(dǎo)致的。
- 確保你的 MyEvent 類是線程安全的，因?yàn)樗赡鼙欢鄠€(gè)線程同時(shí)訪問。
如果對(duì)@Async默認(rèn)線程池感興趣的同學(xué)可以通過這幾個(gè)類進(jìn)行進(jìn)一步查看
- 處理基礎(chǔ)類AsyncExecutionAspectSupport,定義了默認(rèn)執(zhí)行器
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
// 省略其他代碼
return (Executor)beanFactory.getBean(TaskExecutor.class);
}- 異步執(zhí)行器配置類TaskExecutionAutoConfiguration
@Bean
@ConditionalOnMissingBean
public TaskExecutorBuilder taskExecutorBuilder(TaskExecutionProperties properties,
ObjectProvider<TaskExecutorCustomizer> taskExecutorCustomizers,
ObjectProvider<TaskDecorator> taskDecorator) {
TaskExecutionProperties.Pool pool = properties.getPool();
TaskExecutorBuilder builder = new TaskExecutorBuilder();
// 省略其他代碼
return builder;
}
@Lazy
@Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,
AsyncAnnotationBeanPostProcessor.DEFAULT_TASK_EXECUTOR_BEAN_NAME })
@ConditionalOnMissingBean(Executor.class)
public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
return builder.build();
}配置時(shí)間廣播器multicaster
首選創(chuàng)建一個(gè)配置類。
@Configuration
public class AsynchronousSpringEventsConfig {
@Bean(name = "applicationEventMulticaster")
public ApplicationEventMulticaster simpleApplicationEventMulticaster() {
SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
ThreadPoolExecutor e = new ThreadPoolExecutor(10, 20, 10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(10),
new ThreadFactoryBuilder().setNameFormat("notify-onboard-thread-%d").build(),
new ThreadPoolExecutor.AbortPolicy());
eventMulticaster.setTaskExecutor(e);
// 設(shè)置錯(cuò)誤處理器
eventMulticaster.setErrorHandler(new ErrorHandler() {
@Override
public void handleError(Throwable t) {
log.error("get error when notify onboard service", t);
}
});
return eventMulticaster;
}
}然后事件監(jiān)聽器的代碼不用修改，發(fā)布時(shí)間方式也不用修改。
使用這種方式有幾個(gè)注意點(diǎn)
- 所有有的事件監(jiān)聽器都變?yōu)楫惒綀?zhí)行。如果某些監(jiān)聽器需要同步執(zhí)行，你可能需要額外的配置或使用不同的多播器。
- 這里廣播器的名字一定要叫applicationEventMulticaster
// APPLICATION_EVENT_MULTICASTER_BEAN_NAME = applicationEventMulticaster
protected void initApplicationEventMulticaster() {
ConfigurableListableBeanFactory beanFactory = getBeanFactory();
if (beanFactory.containsLocalBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME)) {
this.applicationEventMulticaster =
beanFactory.getBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class);
}
else {
this.applicationEventMulticaster = new SimpleApplicationEventMulticaster(beanFactory);
beanFactory.registerSingleton(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, this.applicationEventMulticaster);
}
}這里為什么要指定errorHandler呢？ 是因?yàn)槿绻恢付ǖ脑?，然后?zhí)行的業(yè)務(wù)邏輯又拋出了異常，這個(gè)時(shí)候異常是會(huì)往外拋的，所以我們就會(huì)在執(zhí)行的方法上加上try-catch, 而有了handler,我們就可以統(tǒng)一處理這個(gè)異常問題。
@Override
public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {
ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
Executor executor = getTaskExecutor();
for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {
if (executor != null) {
executor.execute(() -> invokeListener(listener, event));
}
else {
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);
}
}從 SimpleApplicationEventMulticaster中可以看出來，如果不指定Executor, 則最終事件的執(zhí)行是由同一 個(gè)線程按順序來完成的，同時(shí)任何一個(gè)報(bào)錯(cuò)，都會(huì)導(dǎo)致后續(xù)的監(jiān)聽器執(zhí)行不了(如果存在多個(gè)監(jiān)聽器)。
總結(jié)
至此，我總結(jié)了兩種事件異步執(zhí)行的方式，我讓我同事選擇一種，他選了第二種，如果是你你會(huì)選擇哪一種？
