初识@EnableAsync @Async

清疚 2022-10-12 04:23 282阅读 0赞

项目中常会使用到异步任务去完成耗时的业务逻辑处理，今天使用的是基于springboot的@EnableAsync @Async实现异步任务。

### 1.@Async ###

/**
 * Annotation that marks a method as a candidate for asynchronous execution.
 * Can also be used at the type level, in which case all of the type's methods are
 * considered as asynchronous.
 *
 * In terms of target method signatures, any parameter types are supported.
 * However, the return type is constrained to either {@code void} or
 * {@link java.util.concurrent.Future}. In the latter case, you may declare the
 * more specific {@link org.springframework.util.concurrent.ListenableFuture} or
 * {@link java.util.concurrent.CompletableFuture} types which allow for richer
 * interaction with the asynchronous task and for immediate composition with
 * further processing steps.
 *
 * A {@code Future} handle returned from the proxy will be an actual asynchronous
 * {@code Future} that can be used to track the result of the asynchronous method
 * execution. However, since the target method needs to implement the same signature,
 * it will have to return a temporary {@code Future} handle that just passes a value
 * through: e.g. Spring's {@link AsyncResult}, EJB 3.1's {@link javax.ejb.AsyncResult},
 * or {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}.
 *
 * @author Juergen Hoeller
 * @author Chris Beams
 * @since 3.0
 * @see AnnotationAsyncExecutionInterceptor
 * @see AsyncAnnotationAdvisor
 */
 @Target({ElementType.METHOD, ElementType.TYPE})
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 public @interface Async {
 ......

翻译：

> \*标记一个方法作为异步执行的候选方法的注释。 
> \*也可以在类型级别使用，在这种情况下，类型的所有方法都是 
> 被认为是异步的。 
> ＊ 
> \*  
> 对于目标方法签名，支持任何参数类型。 
> 然而，返回类型被限制为\{@code void\}或 
> \* \{@link java.util.concurrent.Future\}。在后一种情况下，可以声明 
> \*更具体的\{@link org.springframework.util.concurrent。ListenableFuture\}或 
> \* \{@link java . util . concurrent。CompletableFuture\}类型允许更丰富 
> \*与异步任务的交互和与即时复合 
> \*进一步的处理步骤。 
> ＊ 
> 从代理返回的\{@code Future\}句柄将是一个实际的异步 
> \* \{@code Future\}，可以用来跟踪异步方法的结果 
> \*执行。但是，由于目标方法需要实现相同的签名， 
> \*它必须返回一个临时的\{@code Future\}句柄，只传递一个值 
> \*通过:例如Spring的\{@link AsyncResult\}， EJB 3.1的\{@link javax.ejb.AsyncResult\}， 
> \*或\{@link java.util.concurrent.CompletableFuture\#completedFuture(Object)\}。

由此我们可以知道

1.@Async注解在类上说明这个类里的方法全是异步方法，如果只注解在某个方法上则是指明该方法是异步方法。

2.可以不带返回值，

/**
         * 最简单的异步调用，返回值为void
         */
        @Async("taskExecutor")
        public void asyncInvokeSimplest() {
            log.info("asyncSimplest");
        }

3.可以带返回值

/**
 * 异常调用返回Future
 *
 * @param i
 * @return
 */
 @Async("taskExecutor")
 public Future<String> asyncInvokeReturnFuture(int i) {
 log.info("asyncInvokeReturnFuture, parementer={}", i);
 Future<String> future;
 try {
 Thread.sleep(1000 * 1);
 future = new AsyncResult<String>("success:" + i);
 } catch (InterruptedException e) {
 future = new AsyncResult<String>("error");
 }
 return future;
 }

### 2.@EnableAsync ###

/**
 * Enables Spring's asynchronous method execution capability, similar to functionality
 * found in Spring's {@code <task:*>} XML namespace.
 *
 * To be used together with @{@link Configuration Configuration} classes as follows,
 * enabling annotation-driven async processing for an entire Spring application context:
 *
 * <pre class="code">
 * @Configuration
 * @EnableAsync
 * public class AppConfig {
 *
 * }</pre>
 *
 * {@code MyAsyncBean} is a user-defined type with one or more methods annotated with
 * either Spring's {@code @Async} annotation, the EJB 3.1 {@code @javax.ejb.Asynchronous}
 * annotation, or any custom annotation specified via the {@link #annotation} attribute.
 * The aspect is added transparently for any registered bean, for instance via this
 * configuration:
 *
 * <pre class="code">
 * @Configuration
 * public class AnotherAppConfig {
 *
 * @Bean
 * public MyAsyncBean asyncBean() {
 * return new MyAsyncBean();
 * }
 * }</pre>
 *
 * By default, Spring will be searching for an associated thread pool definition:
 * either a unique {@link org.springframework.core.task.TaskExecutor} bean in the context,
 * or an {@link java.util.concurrent.Executor} bean named "taskExecutor" otherwise. If
 * neither of the two is resolvable, a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
 * will be used to process async method invocations. Besides, annotated methods having a
 * {@code void} return type cannot transmit any exception back to the caller. By default,
 * such uncaught exceptions are only logged.
 *
 * To customize all this, implement {@link AsyncConfigurer} and provide:
 * <ul>
 * <li>your own {@link java.util.concurrent.Executor Executor} through the
 * {@link AsyncConfigurer#getAsyncExecutor getAsyncExecutor()} method, and</li>
 * <li>your own {@link org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler
 * AsyncUncaughtExceptionHandler} through the {@link AsyncConfigurer#getAsyncUncaughtExceptionHandler
 * getAsyncUncaughtExceptionHandler()}
 * method.</li>
 * </ul>
 *
 * NOTE: {@link AsyncConfigurer} configuration classes get initialized early
 * in the application context bootstrap. If you need any dependencies on other beans
 * there, make sure to declare them 'lazy' as far as possible in order to let them
 * go through other post-processors as well.
 *
 * <pre class="code">
 * @Configuration
 * @EnableAsync
 * public class AppConfig implements AsyncConfigurer {
 *
 * @Override
 * public Executor getAsyncExecutor() {
 * ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 * executor.setCorePoolSize(7);
 * executor.setMaxPoolSize(42);
 * executor.setQueueCapacity(11);
 * executor.setThreadNamePrefix("MyExecutor-");
 * executor.initialize();
 * return executor;
 * }
 *
 * @Override
 * public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
 * return new MyAsyncUncaughtExceptionHandler();
 * }
 * }</pre>
 *
 * If only one item needs to be customized, {@code null} can be returned to
 * keep the default settings. Consider also extending from {@link AsyncConfigurerSupport}
 * when possible.
 *
 * Note: In the above example the {@code ThreadPoolTaskExecutor} is not a fully managed
 * Spring bean. Add the {@code @Bean} annotation to the {@code getAsyncExecutor()} method
 * if you want a fully managed bean. In such circumstances it is no longer necessary to
 * manually call the {@code executor.initialize()} method as this will be invoked
 * automatically when the bean is initialized.
 *
 * For reference, the example above can be compared to the following Spring XML
 * configuration:
 *
 * <pre class="code">
 * <beans>
 *
 * <task:annotation-driven executor="myExecutor" exception-handler="exceptionHandler"/>
 *
 * <task:executor id="myExecutor" pool-size="7-42" queue-capacity="11"/>
 *
 * <bean id="asyncBean" class="com.foo.MyAsyncBean"/>
 *
 * <bean id="exceptionHandler" class="com.foo.MyAsyncUncaughtExceptionHandler"/>
 *
 * </beans>
 * </pre>
 *
 * The above XML-based and JavaConfig-based examples are equivalent except for the
 * setting of the thread name prefix of the {@code Executor}; this is because
 * the {@code <task:executor>} element does not expose such an attribute. This
 * demonstrates how the JavaConfig-based approach allows for maximum configurability
 * through direct access to actual componentry.
 *
 * The {@link #mode} attribute controls how advice is applied: If the mode is
 * {@link AdviceMode#PROXY} (the default), then the other attributes control the behavior
 * of the proxying. Please note that proxy mode allows for interception of calls through
 * the proxy only; local calls within the same class cannot get intercepted that way.
 *
 * Note that if the {@linkplain #mode} is set to {@link AdviceMode#ASPECTJ}, then the
 * value of the {@link #proxyTargetClass} attribute will be ignored. Note also that in
 * this case the {@code spring-aspects} module JAR must be present on the classpath, with
 * compile-time weaving or load-time weaving applying the aspect to the affected classes.
 * There is no proxy involved in such a scenario; local calls will be intercepted as well.
 *
 * @author Chris Beams
 * @author Juergen Hoeller
 * @author Stephane Nicoll
 * @author Sam Brannen
 * @since 3.1
 * @see Async
 * @see AsyncConfigurer
 * @see AsyncConfigurationSelector
 */
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @Import(AsyncConfigurationSelector.class)
 public @interface EnableAsync {
 ......

翻译：

> ／\*\*
> 
> \*启用Spring的异步方法执行能力，类似于功能
> 
> \*在Spring的\{@code \} XML命名空间中找到。 
> 与@\{@link Configuration Configuration\}类一起使用，如下:
> 
> \*为整个Spring应用上下文启用注解驱动的异步处理:
> 
>  \* <pre class="code"> 
>  \* &\#064;Configuration 
>  \* &\#064;EnableAsync 
>  \* public class AppConfig \{ 
>  \* 
>  \* \}</pre>
> 
> \* \{@code MyAsyncBean\}是一个用户定义的类型，带有一个或多个注释的方法
> 
> \* Spring的\{@code @Async\}注释，EJB 3.1 \{@code @javax.ejb.Asynchronous\}
> 
> 注释，或者任何通过\{@link \#annotation\}属性指定的自定义注释。
> 
> \*对于任何已注册的bean来说，方面都是透明的，例如通过这个
> 
> \*配置:
> 
> ＊ 
> \* <pre class="code"> 
>  \* &\#064;Configuration 
>  \* public class AnotherAppConfig \{ 
>  \* 
>  \*     &\#064;Bean 
>  \*     public MyAsyncBean asyncBean() \{ 
>  \*         return new MyAsyncBean(); 
>  \*     \} 
>  \* \}</pre>
> 
> ＊
> 
> 默认情况下，Spring会搜索一个关联的线程池定义:
> 
> \*要么是一个唯一的\{@link org.springframework.core.task。TaskExecutor\} bean在上下文中，
> 
> \*或\{@link java.util.concurrent。否则，bean命名为“taskExecutor”。如果
> 
> \*这两个都不是可解析的，一个 @link org.springframework.core.task.SimpleAsyncTaskExecutor\}
> 
> \*将用于处理异步方法调用。此外，具有
> 
> \* \{@code void\}返回类型不能将任何异常传递回调用者。默认情况下,
> 
> 这种未捕获的异常只被记录。
> 
> \* 
> 要定制所有这些，实现\{@link AsyncConfigurer\}并提供:
> 
> \* < ul > 
>  <li> your own \{@link java.util.concurrent.Executor Executor\}通过 
> \* \{@link AsyncConfigurer\#getAsyncExecutor getAsyncExecutor()\}方法，和 
> your own \{@link org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler 
> \* AsyncUncaughtExceptionHandler\}通过\{@link AsyncConfigurer\#getAsyncUncaughtExceptionHandler
> 
> \* getAsyncUncaughtExceptionHandler ()\}
> 
> \*方法。< /li>
> 
> \*</ul>
> 
> 说明:\{@link AsyncConfigurer\}配置类会提前初始化
> 
> \*在应用程序上下文引导。如果您需要其他bean的任何依赖项
> 
> 在那里，确保尽可能地声明他们是“懒惰的”，以便让他们还有其他的后处理器> .
> 
>  \* <pre class="code"> 
>  \* &\#064;Configuration 
>  \* &\#064;EnableAsync 
>  \* public class AppConfig implements AsyncConfigurer \{ 
>  \* 
>  \*     &\#064;Override 
>  \*     public Executor getAsyncExecutor() \{ 
>  \*         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
>  \*         executor.setCorePoolSize(7); 
>  \*         executor.setMaxPoolSize(42); 
>  \*         executor.setQueueCapacity(11); 
>  \*         executor.setThreadNamePrefix("MyExecutor-"); 
>  \*         executor.initialize(); 
>  \*         return executor; 
>  \*     \} 
>  \* 
>  \*     &\#064;Override 
>  \*     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() \{ 
>  \*         return new MyAsyncUncaughtExceptionHandler(); 
>  \*     \} 
>  \* \}</pre> 
>  \* 
> ＊
> 
> 如果只需要定制一个项目，\{@code null\}可以返回
> 
> \*保持默认设置。还可以考虑从\{@link AsyncConfigurerSupport\}扩展
> 
> \*如果可能的话。
> 
> 注意:在上面的例子中\{@code ThreadPoolTaskExecutor\}不是完全被管理的
> 
> \* Spring bean。将\{@code @Bean\}注释添加到\{@code getAsyncExecutor()\}方法中
> 
> \*如果您想要一个完全管理的bean。在这种情况下，就没有必要
> 
> \*手动调用\{@code executor.initialize()\}方法，因为它将被调用
> 
> 当bean初始化时自动。
> 
> 作为参考，上面的例子可以与下面的Spring XML相比较
> 
> \*配置:
> 
>  \* <pre class="code"> 
>  \* <beans> 
>  \* 
>  \*     <task:annotation-driven executor="myExecutor" exception-handler="exceptionHandler"/> 
>  \* 
>  \*     <task:executor id="myExecutor" pool-size="7-42" queue-capacity="11"/> 
>  \* 
>  \*     <bean id="asyncBean" class="com.foo.MyAsyncBean"/> 
>  \* 
>  \*     <bean id="exceptionHandler" class="com.foo.MyAsyncUncaughtExceptionHandler"/> 
>  \* 
>  \* </beans> 
>  \* </pre>
> 
> \*以上基于xml和基于javaconfig的例子是等价的，除了
> 
> \*设置\{@code Executor\}的线程名前缀;这是因为
> 
> \{@code \}元素不会公开这样的属性。这
> 
> \*演示了基于javaconfig的方法如何实现最大的可配置性
> 
> \*通过直接访问实际的组件。
> 
> 
> \{@link \#mode\}属性控制如何应用通知:如果模式是
> 
> \* \{@link AdviceMode\#PROXY\}(默认)，然后其他属性控制行为
> 
> \*的代理。请注意代理模式允许截取通过的呼叫
> 
> \*仅代理;同一个类中的本地调用不能以这种方式被拦截。
> 
> ＊ 
> 注意，如果\{@linkplain \#mode\}被设置为\{@link AdviceMode\#ASPECTJ\}，那么
> 
> \{@link \#proxyTargetClass\}属性的值将被忽略。请注意
> 
> \*这种情况下\{@code spring-aspects\}模块JAR必须出现在类路径中
> 
> 将方面应用于受影响的类的编译时编织或加载时编织。
> 
> \*没有代理人参与该等方案;本地调用也会被拦截。

通过以上的解释来看，如果我们没有自定义线程池的话，将会默认使用SimpleAsyncTaskExecutor

### 3.SimpleAsyncTaskExecutor ###

/**
 * {@link TaskExecutor} implementation that fires up a new Thread for each task,
 * executing it asynchronously.
 *
 * Supports limiting concurrent threads through the "concurrencyLimit"
 * bean property. By default, the number of concurrent threads is unlimited.
 *
 * NOTE: This implementation does not reuse threads! Consider a
 * thread-pooling TaskExecutor implementation instead, in particular for
 * executing a large number of short-lived tasks.
 *
 * @author Juergen Hoeller
 * @since 2.0
 * @see #setConcurrencyLimit
 * @see SyncTaskExecutor
 * @see org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
 * @see org.springframework.scheduling.commonj.WorkManagerTaskExecutor
 */
 @SuppressWarnings("serial")
 public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator
 		implements AsyncListenableTaskExecutor, Serializable {
 ......

翻译：

> \* \{@link TaskExecutor\}实现为每个任务触发一个新的线程，
> 
> \*异步执行。
> 
> 支持通过"concurrencyLimit"限制并发线程
> 
> \* bean属性。默认情况下，并发线程的数量是无限制的。
> 
> 注意:此实现不重用线程!考虑
> 
> \*线程池TaskExecutor实现，特别是
> 
> \*执行大量的短期任务。

3.1).SimpleAsyncTaskExecutor提供了限流机制，通过concurrencyLimit属性来控制开关，当concurrencyLimit>=0时开启限流机制，默认关闭限流机制即concurrencyLimit=-1，当关闭情况下，会不断创建新的线程来处理任务，核心代码如下：（异步执行用户任务的SimpleAsyncTaskExecutor。每次执行客户提交给它的任务时，它会启动新的线程，并允许开发者控制并发线程的上限（concurrencyLimit），从而起到一定的资源节流作用。默认时，concurrencyLimit取值为-1，即不启用资源节流。）

//1.是否开启限流 否则不开启限流处理
    //2.执行开始之前检测是否可以满足要求 当前数量++
    //3.开启限流将执行的Runable进行封装，执行完成调用final方法 当前数量--
    public void execute(Runnable task, long startTimeout) {
       Assert.notNull(task, "Runnable must not be null");
       Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
       //判断是否开启限流机制
       if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
          //执行前置操作，进行限流
          this.concurrencyThrottle.beforeAccess();
          //执行完线程任务，会执行后置操作concurrencyThrottle.afterAccess()，配合进行限流
          doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
       }
       else {
          doExecute(taskToUse);
       }
    }

开启节流配置：

@Configuration
    @EnableAsync
    public class AsyncCommonConfig extends AsyncConfigurerSupport {
        @Override
        public Executor getAsyncExecutor() {
            SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
            //设置允许同时执行的线程数为10
     executor.setConcurrencyLimit(10);
            return executor;
        }
    }

3.2).SimpleAsyncTaskExecutor限流实现  
首先任务进来，会循环判断当前执行线程数是否超过concurrencyLimit，如果超了，则当前线程调用wait方法，释放monitor对象锁，进入等待

protected void beforeAccess() {
    	if (this.concurrencyLimit == NO_CONCURRENCY) {
    		throw new IllegalStateException(
    				"Currently no invocations allowed - concurrency limit set to NO_CONCURRENCY");
    	}
    	if (this.concurrencyLimit > 0) {
    		boolean debug = logger.isDebugEnabled();
    		synchronized (this.monitor) {
    			boolean interrupted = false;
    			while (this.concurrencyCount >= this.concurrencyLimit) {
    				if (interrupted) {
    					throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
    							"but concurrency limit still does not allow for entering");
    				}
    				if (debug) {
    					logger.debug("Concurrency count " + this.concurrencyCount +
    							" has reached limit " + this.concurrencyLimit + " - blocking");
    				}
    				try {
    					this.monitor.wait();
    				}
    				catch (InterruptedException ex) {
    					// Re-interrupt current thread, to allow other threads to react.
    					Thread.currentThread().interrupt();
    					interrupted = true;
    				}
    			}
    			if (debug) {
    				logger.debug("Entering throttle at concurrency count " + this.concurrencyCount);
    			}
    			this.concurrencyCount++;
    		}
    	}
    }

3.3)线程任务执行完毕后，当前执行线程数会减一，会调用monitor对象的notify方法，唤醒等待状态下的线程，等待状态下的线程会竞争monitor锁，竞争到，会继续执行线程任务。

protected void afterAccess() {
    	if (this.concurrencyLimit >= 0) {
    		synchronized (this.monitor) {
    			this.concurrencyCount--;
    			if (logger.isDebugEnabled()) {
    				logger.debug("Returning from throttle at concurrency count " + this.concurrencyCount);
    			}
    			this.monitor.notify();
    		}
    	}
    }

**看了源码了解了SimpleAsyncTaskExecutor有限流机制.**

> 1.开启限流情况下，能有效控制应用线程数  
> 2.虽然可以有效控制线程数，但执行效率会降低，会出现主线程等待，线程竞争的情况。  
> 3.限流机制适用于任务处理比较快的场景，对于应用处理时间比较慢的场景并不适用。==

最终解决办法：  
1.自定义线程池，使用LinkedBlockingQueue阻塞队列来限定线程池的上限  
2.定义拒绝策略，如果队列满了，则拒绝处理该任务，打印日志，代码如下：

public class AsyncConfig implements AsyncConfigurer {
        private Logger logger = LogManager.getLogger();
    
        @Value("${thread.pool.corePoolSize:10}")
        private int corePoolSize;
    
        @Value("${thread.pool.maxPoolSize:20}")
        private int maxPoolSize;
    
        @Value("${thread.pool.keepAliveSeconds:4}")
        private int keepAliveSeconds;
    
        @Value("${thread.pool.queueCapacity:512}")
        private int queueCapacity;
    
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(corePoolSize);
            executor.setMaxPoolSize(maxPoolSize);
            executor.setKeepAliveSeconds(keepAliveSeconds);
            executor.setQueueCapacity(queueCapacity);
            executor.setRejectedExecutionHandler((Runnable r, ThreadPoolExecutor exe) -> {
                    logger.warn("当前任务线程池队列已满.");
            });
            executor.initialize();
            return executor;
        }
    
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new AsyncUncaughtExceptionHandler() {
                @Override
                public void handleUncaughtException(Throwable ex , Method method , Object... params) {
                    logger.error("线程池执行任务发生未知异常.", ex);
                }
            };
        }
    }