关于Spring中的@Async注解以及为什么不建议使用 - Java技术债务

@Async
public CompletableFuture doSomethingAsync() {    
	// 异步执行一些操作
}

@Configuration
@Slf4j
public class AsyncConfiguration implements AsyncConfigurer {

    /**
     * 核心线程
     */
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() == 1 ? 4 : Runtime.getRuntime().availableProcessors();

    @Primary
    @Bean("commonExecutor")
    public Executor commonExecutor() {
        log.info("==== start initialize common async executor =====");
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(CORE_POOL_SIZE);
        taskExecutor.setQueueCapacity(1024);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix("commonExecutor--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(20);
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Bean("xmarsRecommendationExecutor")
    public Executor xmarsRecommendationExecutor() {
        log.info("==== start initialize xmars recommendation async executor =====");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(CORE_POOL_SIZE);
        executor.setQueueCapacity(10240);
        executor.setKeepAliveSeconds(120);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setThreadNamePrefix("xmars-recommendation-executor--");
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.initialize();
        return executor;
    }

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

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return ((ex, method, params) -> log.error("执行异步任务:{}", method, ex));
    }
}

// Spring boot启用：
@EnableAsync
@EnableTransactionManagement
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

/**
 * 带参数的异步调用 异步方法可以传入参数    
 *  对于返回值是void，异常会被AsyncUncaughtExceptionHandler处理掉    

 * @param s
 */
@Async
public void asyncInvokeWithException(String s) {
    log.info("asyncInvokeWithParameter, parementer={}", s);
    throw new IllegalArgumentException(s);
}

/**
 * 异常调用返回Future
 * 对于返回值是Future，不会被AsyncUncaughtExceptionHandler处理，需要我们在方法中捕获异常并处理
 * 或者在调用方在调用Futrue.get时捕获异常进行处理
 *
 * @param i
 * @return
 */
@Async
public Future asyncInvokeReturnFuture(int i) {
    log.info("asyncInvokeReturnFuture, parementer={}", i);
    Future future;
    try {
        Thread.sleep(1000 * 1);
        future = new AsyncResult("success:" + i);
        throw new IllegalArgumentException("a");
    } catch (InterruptedException e) {
        future = new AsyncResult("error");
    } catch (IllegalArgumentException e) {
        future = new AsyncResult("error-IllegalArgumentException");
    }
    return future;
}

Executor.class:ThreadPoolExecutorAdapter->ThreadPoolExecutor->AbstractExecutorService->ExecutorService->Executor

TaskExecutor.class:ThreadPoolTaskExecutor->SchedulingTaskExecutor->AsyncTaskExecutor->TaskExecutor

@EnableAsync
@Configuration
public class TaskPoolConfig {
   @Bean(name = AsyncExecutionAspectSupport.DEFAULT_TASK_EXECUTOR_BEAN_NAME)
   public Executor taskExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
       executor.setCorePoolSize(10);
       //最大线程数
       executor.setMaxPoolSize(20);
       //队列容量
       executor.setQueueCapacity(200);
       //活跃时间
       executor.setKeepAliveSeconds(60);
       //线程名字前缀
       executor.setThreadNamePrefix("taskExecutor-");
       executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
       return executor;
   }
  @Bean(name = "new_task")
   public Executor taskExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
       executor.setCorePoolSize(10);
       //最大线程数
       executor.setMaxPoolSize(20);
       //队列容量
       executor.setQueueCapacity(200);
       //活跃时间
       executor.setKeepAliveSeconds(60);
       //线程名字前缀
       executor.setThreadNamePrefix("taskExecutor-");
       executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
       return executor;
   }
}

// 定义了一个受保护的方法，该方法接受一个Method类型的参数，并返回一个AsyncTaskExecutor类型的对象。  
protected AsyncTaskExecutor determineAsyncExecutor(Method method) {  
    // 尝试从executors这个Map中获取与给定方法关联的AsyncTaskExecutor。  
    AsyncTaskExecutor executor = this.executors.get(method);  
  
    // 如果executor为空，表示当前方法还没有对应的AsyncTaskExecutor。  
    if (executor == null) {  
        // 根据方法确定执行器的标识符。  
        String qualifier = getExecutorQualifier(method);  
  
        // 如果获取到了非空的执行器标识符。  
        if (StringUtils.hasLength(qualifier)) {  
            // 在beanFactory中查找具有指定标识符的执行器。  
            targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);  
        }  
        // 如果没有获取到执行器标识符，则使用默认的执行器。  
        else {  
            targetExecutor = this.defaultExecutor.get();  
        }  
  
        // 如果仍然没有找到执行器，则返回null。  
        if (targetExecutor == null) {  
            return null;  
        }  
  
        // 如果targetExecutor是AsyncListenableTaskExecutor的实例，则直接赋值给executor。  
        // 否则，使用targetExecutor创建一个新的TaskExecutorAdapter并将其赋值给executor。  
        executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?  
                (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));  
  
        // 将方法和执行器的关联存储到executors这个Map中。  
        this.executors.put(method, executor);  
    }  
  
    // 返回与给定方法关联的AsyncTaskExecutor。  
    return executor;  
}

public interface AsyncConfigurer {

	/**
	 * The {@link Executor} instance to be used when processing async
	 * method invocations.
	 */
	@Nullable
	default Executor getAsyncExecutor() {
		return null;
	}

	/**
	 * The {@link AsyncUncaughtExceptionHandler} instance to be used
	 * when an exception is thrown during an asynchronous method execution
	 * with {@code void} return type.
	 */
	@Nullable
	default AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
		return null;
	}
}

// 定义了一个受保护的方法，该方法接受一个可空的BeanFactory作为参数，并返回一个Executor类型的对象。  
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {  
    // 如果beanFactory不为空，则继续执行以下逻辑。  
    if (beanFactory != null) {  
        try {  
            // 尝试从beanFactory中获取一个TaskExecutor类型的bean。  
            // 这里特别指定了TaskExecutor而不是普通的Executor，因为后者可能会匹配到ScheduledExecutorService，  
            // 这对于我们的目的来说是不可用的。TaskExecutor更明确地为此设计。  
            return beanFactory.getBean(TaskExecutor.class);  
        }  
        catch (NoUniqueBeanDefinitionException ex) {  
            // 如果找到了多个TaskExecutor bean，记录一条debug级别的日志。  
            logger.debug("Could not find unique TaskExecutor bean", ex);  
            try {  
                // 尝试获取名为DEFAULT_TASK_EXECUTOR_BEAN_NAME的Executor类型的bean。  
                return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);  
            }  
            catch (NoSuchBeanDefinitionException ex2) {  
                // 如果连默认的TaskExecutor bean都没有找到，并且logger的info级别是启用的，  
                // 则记录一条info级别的日志，说明找到了多个TaskExecutor bean，但没有一个是名为'taskExecutor'的，  
                // 并列出所有找到的bean的名称。  
                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 bean都没有找到，记录一条debug级别的日志。  
            logger.debug("Could not find default TaskExecutor bean", ex);  
            try {  
                // 再次尝试获取名为DEFAULT_TASK_EXECUTOR_BEAN_NAME的Executor类型的bean。  
                return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);  
            }  
            catch (NoSuchBeanDefinitionException ex2) {  
                // 如果连名为DEFAULT_TASK_EXECUTOR_BEAN_NAME的bean都没有找到，记录一条info级别的日志，  
                // 说明没有找到用于异步处理的TaskExecutor bean。  
                logger.info("No task executor bean found for async processing: " +  
                        "no bean of type TaskExecutor and no bean named 'taskExecutor' either");  
            }  
            // 如果没有找到任何可用的bean，方法返回null。  
        }  
    }  
    // 如果beanFactory本身就是null，方法也返回null。  
    return null;  
}

/**
 * This implementation searches for a unique {@link org.springframework.core.task.TaskExecutor}
 * bean in the context, or for an {@link Executor} bean named "taskExecutor" otherwise.
 * If neither of the two is resolvable (e.g. if no {@code BeanFactory} was configured at all),
 * this implementation falls back to a newly created {@link SimpleAsyncTaskExecutor} instance
 * for local use if no default could be found.
 * @see #DEFAULT_TASK_EXECUTOR_BEAN_NAME
 */
@Override
@Nullable
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
	Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
	return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
}