java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > Async线程池使用

Async的线程池使用选择解析

作者：我是一颗小虎牙_

这篇文章主要为大家介绍了Async的线程池使用选择解析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪

前言

在Spring中我们经常会用到异步操作，注解中使用 @EnableAsync 和 @Async 就可以使用它了。但是最近发现在异步中线程号使用的是我们项目中自定义的线程池 ThreadPoolTaskExecutor 而不是之前熟悉的 SimpleAsyncTaskExecutor

那么来看一下他的执行过程吧。

正文

public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {
    private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME = "org.springframework.scheduling.aspectj.AspectJAsyncConfiguration";
    public AsyncConfigurationSelector() {
    }
    @Nullable
    public String[] selectImports(AdviceMode adviceMode) {
        switch(adviceMode) {
        case PROXY:
            return new String[]{ProxyAsyncConfiguration.class.getName()};
        case ASPECTJ:
            return new String[]{"org.springframework.scheduling.aspectj.AspectJAsyncConfiguration"};
        default:
            return null;
        }
    }
}

public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {
    public ProxyAsyncConfiguration() {
    }
    @Bean(
        name = {"org.springframework.context.annotation.internalAsyncAnnotationProcessor"}
    )
    @Role(2)
    public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
        Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
        AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
        bpp.configure(this.executor, this.exceptionHandler);
        Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
        if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {
            bpp.setAsyncAnnotationType(customAsyncAnnotation);
        }
        bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
        bpp.setOrder((Integer)this.enableAsync.getNumber("order"));
        return bpp;
    }
}
public void setBeanFactory(BeanFactory beanFactory) {
    super.setBeanFactory(beanFactory);
    //定义一个切面
    AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);
    if (this.asyncAnnotationType != null) {
        advisor.setAsyncAnnotationType(this.asyncAnnotationType);
    }
    advisor.setBeanFactory(beanFactory);
    this.advisor = advisor;
}

    public AsyncAnnotationAdvisor(
            @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {
        Set<Class<? extends Annotation>> asyncAnnotationTypes = new LinkedHashSet<>(2);
        asyncAnnotationTypes.add(Async.class);
        try {
            asyncAnnotationTypes.add((Class<? extends Annotation>)
                    ClassUtils.forName("javax.ejb.Asynchronous", AsyncAnnotationAdvisor.class.getClassLoader()));
        }
        catch (ClassNotFoundException ex) {
            // If EJB 3.1 API not present, simply ignore.
        }
        //通知
        this.advice = buildAdvice(executor, exceptionHandler);
        //切入点
        this.pointcut = buildPointcut(asyncAnnotationTypes);
    }
protected Advice buildAdvice(
            @Nullable Supplier&lt;Executor&gt; executor, @Nullable Supplier&lt;AsyncUncaughtExceptionHandler&gt; exceptionHandler) {
        AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
        //配置拦截器
        interceptor.configure(executor, exceptionHandler);
        return interceptor;
    }

public void configure(@Nullable Supplier&lt;Executor&gt; defaultExecutor,
            @Nullable Supplier&lt;AsyncUncaughtExceptionHandler&gt; exceptionHandler) {
        //默认执行器
        this.defaultExecutor = new SingletonSupplier&lt;&gt;(defaultExecutor, () -&gt; getDefaultExecutor(this.beanFactory));
        this.exceptionHandler = new SingletonSupplier&lt;&gt;(exceptionHandler, SimpleAsyncUncaughtExceptionHandler::new);
    }

/**
 * 父类
 * 获取或构建此通知实例的默认执行器
 * 这里返回的执行器将被缓存以供后续使用
 * 默认实现搜索唯一的TaskExecutor的bean
 * 在上下文中，用于名为“taskExecutor”的Executor bean。
 * 如果两者都不是可解析的，这个实现将返回 null
 */
@Nullable
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
    if (beanFactory != null) {
        try {
            // 搜索唯一的一个TaskExecutor类型的bean并返回
            return beanFactory.getBean(TaskExecutor.class);
        }
        catch (NoUniqueBeanDefinitionException ex) {
            //找不到唯一一个bean异常后，搜索一个TaskExecutor类型的“taskExecutor”的bean并返回
            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) {
            //未找到异常时搜索一个TaskExecutor类型的“taskExecutor”的bean并返回
            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 -&gt; either using local default executor or none at all...
        }
    }
    return null;
}
/**
 * 子类
 * 如父类为null则重新实例化一个名为SimpleAsyncTaskExecutor的执行器
 */
@Override
@Nullable
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
    Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
    return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
}

所以，到了这一步就可以理解为什么异步线程名默认叫 SimpleAsyncTaskExecutor-xx ，为什么有了自己的线程池有可能异步用到了自己的线程池配置。

我们有这个切入点之后，每次请求接口执行异步方法前都会执行 AsyncExecutionInterceptor#invoke() ， determineAsyncExecutor 用来决策使用哪个执行器

@Nullable
protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
    //在缓存的执行器中选择一个对应方法的执行器
    AsyncTaskExecutor executor = (AsyncTaskExecutor)this.executors.get(method);
    if (executor == null) {
        //获取@Async注解中的value（指定的执行器）
        String qualifier = this.getExecutorQualifier(method);
        Executor targetExecutor;
        if (StringUtils.hasLength(qualifier)) {
            //获取指定执行器的bean
            targetExecutor = this.findQualifiedExecutor(this.beanFactory, qualifier);
        } else {
            //选择默认的执行器
            targetExecutor = (Executor)this.defaultExecutor.get();
        }
        if (targetExecutor == null) {
            return null;
        }
        executor = targetExecutor instanceof AsyncListenableTaskExecutor ? (AsyncListenableTaskExecutor)targetExecutor : new TaskExecutorAdapter(targetExecutor);
        //将执行器进行缓存
        this.executors.put(method, executor);
    }
    return (AsyncTaskExecutor)executor;
}

当有了执行器调用 doSubmit 方法将任务加入到执行器中。

异步任务，默认将采用SimpleAsyncTaskExecutor作为执行器！它有如下特点：

不复用线程，也就是说为每个任务新起一个线程。但是可以通过 concurrencyLimit 属性来控制并发线程数量，但是默认情况下不做限制( concurrencyLimit 取值为-1)。
因此，如果我们使用异步任务，一定不能采用默认执行器的配置，以防OOM异常！最好的方式是指定执行器！

总结

本文主要以看源码的方式来了解异步注解 @Async 是如何在项目中选择线程以及使用线程的，尽量给异步任务指定一个独有线程池，这样会的避免不与其他业务共用线程池而造成影响。

以上就是Async的线程池使用选择解析的详细内容，更多关于Async线程池使用的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:
阅读全文