聊聊spring的async注解

本文主要聊聊spring中的async注解。

AsyncConfigurer

@EnableAsync(proxyTargetClass = true)
@Configuration
public class AsyncConfig implements AsyncConfigurer{

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncConfig.class);

    @Override
    public Executor getAsyncExecutor() {
        return null;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncUncaughtExceptionHandler() {
            @Override
            public void handleUncaughtException(Throwable throwable, Method method, Object... params) {
                LOGGER.error(throwable);
            }
        };
    }
}

getAsyncExecutor

这里讨论一下getAsyncExecutor这里定义null的情况。 spring-context-4.3.9.RELEASE-sources.jar!/org/springframework/scheduling/annotation/AbstractAsyncConfiguration.java

@Configuration
public abstract class AbstractAsyncConfiguration implements ImportAware {

    protected AnnotationAttributes enableAsync;

    protected Executor executor;

    protected AsyncUncaughtExceptionHandler exceptionHandler;

    @Override
    public void setImportMetadata(AnnotationMetadata importMetadata) {
        this.enableAsync = AnnotationAttributes.fromMap(
                importMetadata.getAnnotationAttributes(EnableAsync.class.getName(), false));
        if (this.enableAsync == null) {
            throw new IllegalArgumentException(
                    "@EnableAsync is not present on importing class " + importMetadata.getClassName());
        }
    }

    /**
     * Collect any {@link AsyncConfigurer} beans through autowiring.
     */
    @Autowired(required = false)
    void setConfigurers(Collection<AsyncConfigurer> configurers) {
        if (CollectionUtils.isEmpty(configurers)) {
            return;
        }
        if (configurers.size() > 1) {
            throw new IllegalStateException("Only one AsyncConfigurer may exist");
        }
        AsyncConfigurer configurer = configurers.iterator().next();
        this.executor = configurer.getAsyncExecutor();
        this.exceptionHandler = configurer.getAsyncUncaughtExceptionHandler();
    }

}

这里从AsyncConfigurer获取executor

AsyncExecutionInterceptor

spring-aop-4.3.9.RELEASE-sources.jar!/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java

/**
     * Intercept the given method invocation, submit the actual calling of the method to
     * the correct task executor and return immediately to the caller.
     * @param invocation the method to intercept and make asynchronous
     * @return {@link Future} if the original method returns {@code Future}; {@code null}
     * otherwise.
     */
    @Override
    public Object invoke(final MethodInvocation invocation) throws Throwable {
        Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
        Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
        final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);

        AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
        if (executor == null) {
            throw new IllegalStateException(
                    "No executor specified and no default executor set on AsyncExecutionInterceptor either");
        }

        Callable<Object> task = new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                try {
                    Object result = invocation.proceed();
                    if (result instanceof Future) {
                        return ((Future<?>) result).get();
                    }
                }
                catch (ExecutionException ex) {
                    handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
                }
                catch (Throwable ex) {
                    handleError(ex, userDeclaredMethod, invocation.getArguments());
                }
                return null;
            }
        };

        return doSubmit(task, executor, invocation.getMethod().getReturnType());
    }

AsyncExecutionAspectSupport.determineAsyncExecutor

spring-aop-4.3.9.RELEASE-sources.jar!/org/springframework/aop/interceptor/AsyncExecutionAspectSupport.java

**
     * Determine the specific executor to use when executing the given method.
     * Should preferably return an {@link AsyncListenableTaskExecutor} implementation.
     * @return the executor to use (or {@code null}, but just if no default executor is available)
     */
    protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
        AsyncTaskExecutor executor = this.executors.get(method);
        if (executor == null) {
            Executor targetExecutor;
            String qualifier = getExecutorQualifier(method);
            if (StringUtils.hasLength(qualifier)) {
                targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);
            }
            else {
                targetExecutor = this.defaultExecutor;
                if (targetExecutor == null) {
                    synchronized (this.executors) {
                        if (this.defaultExecutor == null) {
                            this.defaultExecutor = getDefaultExecutor(this.beanFactory);
                        }
                        targetExecutor = this.defaultExecutor;
                    }
                }
            }
            if (targetExecutor == null) {
                return null;
            }
            executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?
                    (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));
            this.executors.put(method, executor);
        }
        return executor;
    }

@Aync注解有个value可以标注使用哪个executor,这里的getExecutorQualifier就是寻找这个标识。 这里如果defaultExecutor为null的话,则获取找默认的executor

/**
     * Retrieve or build a default executor for this advice instance.
     * An executor returned from here will be cached for further use.
     * <p>The default implementation searches for a unique {@link TaskExecutor} bean
     * in the context, or for an {@link Executor} bean named "taskExecutor" otherwise.
     * If neither of the two is resolvable, this implementation will return {@code null}.
     * @param beanFactory the BeanFactory to use for a default executor lookup
     * @return the default executor, or {@code null} if none available
     * @since 4.2.6
     * @see #findQualifiedExecutor(BeanFactory, String)
     * @see #DEFAULT_TASK_EXECUTOR_BEAN_NAME
     */
    protected Executor getDefaultExecutor(BeanFactory beanFactory) {
        if (beanFactory != null) {
            try {
                // Search for TaskExecutor bean... not plain Executor since that would
                // match with ScheduledExecutorService as well, which is unusable for
                // our purposes here. TaskExecutor is more clearly designed for it.
                return beanFactory.getBean(TaskExecutor.class);
            }
            catch (NoUniqueBeanDefinitionException ex) {
                logger.debug("Could not find unique TaskExecutor bean", ex);
                try {
                    return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);
                }
                catch (NoSuchBeanDefinitionException ex2) {
                    if (logger.isInfoEnabled()) {
                        logger.info("More than one TaskExecutor bean found within the context, and none is named " +
                                "'taskExecutor'. Mark one of them as primary or name it 'taskExecutor' (possibly " +
                                "as an alias) in order to use it for async processing: " + ex.getBeanNamesFound());
                    }
                }
            }
            catch (NoSuchBeanDefinitionException ex) {
                logger.debug("Could not find default TaskExecutor bean", ex);
                try {
                    return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);
                }
                catch (NoSuchBeanDefinitionException ex2) {
                    logger.info("No task executor bean found for async processing: " +
                            "no bean of type TaskExecutor and no bean named 'taskExecutor' either");
                }
                // Giving up -> either using local default executor or none at all...
            }
        }
        return null;
    }

如果工程里头没有定义默认的task executor的话,则获取bean的时候会抛出NoSuchBeanDefinitionException

AsyncExecutionInterceptor.getDefaultExecutor

spring-aop-4.3.9.RELEASE-sources.jar!/org/springframework/aop/interceptor/AsyncExecutionInterceptor.java

protected Executor getDefaultExecutor(BeanFactory beanFactory) {
        Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
        return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
    }

AsyncExecutionInterceptor重写了getDefaultExecutor方法,先调用AsyncExecutionAspectSupport的getDefaultExecutor,如果默认的找不到,这里new一个SimpleAsyncTaskExecutor

Executor关闭问题

如果是在AsyncConfigurer定义的executor,没有受spring托管,貌似是不会在spring context关闭的时候主动shutdown,这个可能是个问题。

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport
        implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {
        //...
}

spring-context-4.3.9.RELEASE-sources.jar!/org/springframework/scheduling/concurrent/ExecutorConfigurationSupport.java

public abstract class ExecutorConfigurationSupport extends CustomizableThreadFactory
        implements BeanNameAware, InitializingBean, DisposableBean {

    //...
    /**
     * Perform a shutdown on the underlying ExecutorService.
     * @see java.util.concurrent.ExecutorService#shutdown()
     * @see java.util.concurrent.ExecutorService#shutdownNow()
     * @see #awaitTerminationIfNecessary()
     */
    public void shutdown() {
        if (logger.isInfoEnabled()) {
            logger.info("Shutting down ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
        }
        if (this.waitForTasksToCompleteOnShutdown) {
            this.executor.shutdown();
        }
        else {
            this.executor.shutdownNow();
        }
        awaitTerminationIfNecessary();
    }

    /**
     * Wait for the executor to terminate, according to the value of the
     * {@link #setAwaitTerminationSeconds "awaitTerminationSeconds"} property.
     */
    private void awaitTerminationIfNecessary() {
        if (this.awaitTerminationSeconds > 0) {
            try {
                if (!this.executor.awaitTermination(this.awaitTerminationSeconds, TimeUnit.SECONDS)) {
                    if (logger.isWarnEnabled()) {
                        logger.warn("Timed out while waiting for executor" +
                                (this.beanName != null ? " '" + this.beanName + "'" : "") + " to terminate");
                    }
                }
            }
            catch (InterruptedException ex) {
                if (logger.isWarnEnabled()) {
                    logger.warn("Interrupted while waiting for executor" +
                            (this.beanName != null ? " '" + this.beanName + "'" : "") + " to terminate");
                }
                Thread.currentThread().interrupt();
            }
        }
    }
}

ExecutorConfigurationSupport实现了DisposableBean接口,重写了destory方法,在里头调用shutdown

因此,最好将ThreadPoolTaskExecutor的定义托管给spring,这样可以优化关闭。

小结

async注解没有指定executor

  • 如果AsyncConfigurer没有定义executor,则会去寻找spring托管的名为taskExecutor的executor,如果没有,则抛出NoSuchBeanDefinitionException,返回null,然后由AsyncExecutionInterceptor.getDefaultExecutor去new一个SimpleAsyncTaskExecutor,不过这个不是spring托管的
  • 如果AsyncConfigurer定义了executor,则这个也不是spring托管的 不是spring托管的executor,需要自己额外去监听事件,然后优雅关闭

async注解指定executor

比如

@Async("myTaskExecutor")
public void xxxx(){

}

这个则使用指定的myTaskExecutor,而不是AsyncConfigurer中定义的executor。

推荐async注解指定task executor,然后AsyncConfigurer的getAsyncExecutor返回null,让它去寻找默认的taskExecutor(自己应用里头都默认定义一个taskExecutor给spring托管)

doc

  • Task Execution and Scheduling

原文发布于微信公众号 - 码匠的流水账(geek_luandun)

原文发表时间:2017-09-25

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏mukekeheart的iOS之旅

No.013 Roman to Integer

13. Roman to Integer Total Accepted: 95998 Total Submissions: 234087 Difficulty:...

24950
来自专栏码匠的流水账

聊聊HystrixEventNotifier

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/strategy/eventnotifier/Hyst...

12820
来自专栏码匠的流水账

聊聊sentinel的NettyHttpCommandCenter

com/alibaba/csp/sentinel/transport/command/NettyHttpCommandCenter.java

18810
来自专栏码匠的流水账

聊聊spring cloud gateway的NettyConfiguration

本文主要研究下spring cloud gateway的NettyConfiguration

42610
来自专栏函数式编程语言及工具

FunDA(11)- 数据库操作的并行运算:Parallel data processing

   FunDA最重要的设计目标之一就是能够实现数据库操作的并行运算。我们先重温一下fs2是如何实现并行运算的。我们用interleave、merge、eith...

20480
来自专栏码匠的流水账

聊聊resilience4j的bulkhead

resilience4j-bulkhead-0.13.0-sources.jar!/io/github/resilience4j/bulkhead/Bulkhe...

21310
来自专栏码匠的流水账

聊聊sentinel的SentinelWebAutoConfiguration

本文主要研究一下sentinel的SentinelWebAutoConfiguration

20700
来自专栏码匠的流水账

聊聊spring cloud的EurekaServerInitializerConfiguration

本文主要研究一下spring cloud的EurekaServerInitializerConfiguration

15620
来自专栏cmazxiaoma的架构师之路

你真的会用HttpMessageConverter吗?

1.7K60
来自专栏码匠的流水账

聊聊spring for kafka对producer的封装与集成

本文主要解析一下spring for apache kafka对原生的kafka client producer的封装与集成。

16520

扫码关注云+社区

领取腾讯云代金券