Spring之異步任務(wù)@Async解讀

更新時(shí)間：2023年07月20日 09:28:24 作者：愛(ài)吃牛肉的大老虎

這篇文章主要介紹了Spring之異步任務(wù)@Async，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

1 異步@Async詳解

1.1 引言

在java中異步線(xiàn)程很重要，比如在業(yè)務(wù)流處理時(shí)，需要通知硬件設(shè)備，發(fā)短信通知用戶(hù)，或者需要上傳一些圖片資源到其他服務(wù)器這種耗時(shí)的操作，在主線(xiàn)程里處理會(huì)阻塞整理流程，而且我們也不需要等待處理結(jié)果之后再進(jìn)行下一步操作，這時(shí)候就可以使用異步線(xiàn)程進(jìn)行處理，這樣主線(xiàn)程不會(huì)因?yàn)檫@些耗時(shí)的操作而阻塞，保證主線(xiàn)程的流程可以正常進(jìn)行。

最近在項(xiàng)目中使用了很多線(xiàn)程的操作，在這做個(gè)記錄

1.2 異步說(shuō)明和原理

使用地方說(shuō)明：

在方法上使用該@Async注解，申明該方法是一個(gè)異步任務(wù)；
在類(lèi)上面使用該@Async注解，申明該類(lèi)中的所有方法都是異步任務(wù)；
使用此注解的方法的類(lèi)對(duì)象，必須是spring管理下的bean對(duì)象；
要想使用異步任務(wù)，需要在主類(lèi)上開(kāi)啟異步配置，即，配置上@EnableAsync注解；

@Async的原理概括：

@Async的原理是通過(guò) Spring AOP 動(dòng)態(tài)代理的方式來(lái)實(shí)現(xiàn)的。

Spring 容器啟動(dòng)初始化bean時(shí)，判斷類(lèi)中是否使用了@Async注解，如果使用了則為其創(chuàng)建切入點(diǎn)和切入點(diǎn)處理器，根據(jù)切入點(diǎn)創(chuàng)建代理，在線(xiàn)程調(diào)用@Async注解標(biāo)注的方法時(shí)，會(huì)調(diào)用代理，執(zhí)行切入點(diǎn)處理器invoke方法，將方法的執(zhí)行提交給線(xiàn)程池中的另外一個(gè)線(xiàn)程來(lái)處理，從而實(shí)現(xiàn)了異步執(zhí)行。

所以，需要注意的一個(gè)錯(cuò)誤用法是，如果a方法調(diào)用它同類(lèi)中的標(biāo)注@Async的b方法，是不會(huì)異步執(zhí)行的，因?yàn)閺腶方法進(jìn)入調(diào)用的都是該類(lèi)對(duì)象本身，不會(huì)進(jìn)入代理類(lèi)。

因此，相同類(lèi)中的方法調(diào)用帶@Async的方法是無(wú)法異步的，這種情況仍然是同步。

1.3 @Async使用

在Spring中啟用@Async：

@Async注解在使用時(shí)，如果不指定線(xiàn)程池的名稱(chēng)，則使用Spring默認(rèn)的線(xiàn)程池，Spring默認(rèn)的線(xiàn)程池為SimpleAsyncTaskExecutor。
方法上一旦標(biāo)記了這個(gè)@Async注解，當(dāng)其它線(xiàn)程調(diào)用這個(gè)方法時(shí)，就會(huì)開(kāi)啟一個(gè)新的子線(xiàn)程去異步處理該業(yè)務(wù)邏輯。

1.3.1 啟動(dòng)類(lèi)中增加@EnableAsync

以Spring boot為例，啟動(dòng)類(lèi)中增加@EnableAsync：

@EnableAsync
@SpringBootApplication
public class ManageApplication {
    //...
}

1.3.2 方法上加@Async注解

@Component
public class MyAsyncTask {
     @Async
    public void asyncCpsItemImportTask(Long platformId, String jsonList){
        //...具體業(yè)務(wù)邏輯
    }
}

1.4 @Async異步線(xiàn)程池

1.4.1 默認(rèn)線(xiàn)程池

上面的配置會(huì)啟用默認(rèn)的線(xiàn)程池/執(zhí)行器，異步執(zhí)行指定的方法。

Spring默認(rèn)的線(xiàn)程池的默認(rèn)配置：

默認(rèn)核心線(xiàn)程數(shù)：8，
最大線(xiàn)程數(shù)：Integet.MAX_VALUE，
隊(duì)列使用LinkedBlockingQueue，
容量是：Integet.MAX_VALUE，
空閑線(xiàn)程保留時(shí)間：60s，
線(xiàn)程池拒絕策略：AbortPolicy

缺點(diǎn)：從最大線(xiàn)程數(shù)的配置上，相信看到問(wèn)題：并發(fā)情況下，會(huì)無(wú)限創(chuàng)建線(xiàn)程

默認(rèn)線(xiàn)程池的上述缺陷如何解決：答案是，自定義配置參數(shù)就可以了

1.4.2 在配置文件中配置

spring:
  task:
    execution:
      pool:
        max-size: 6
        core-size: 3
        keep-alive: 3s
        queue-capacity: 1000
        thread-name-prefix: name

1.4.3 自定義線(xiàn)程池

在業(yè)務(wù)場(chǎng)景中，有時(shí)需要使用自己定義的執(zhí)行器來(lái)跑異步的業(yè)務(wù)邏輯，那該怎么辦呢？答案是，自定義線(xiàn)程池。

1.4.3.1 編寫(xiě)配置類(lèi)

@Configuration
@Data
public class ExecutorConfig{
    //核心線(xiàn)程
    private int corePoolSize;
    //最大線(xiàn)程
    private int maxPoolSize;
    //隊(duì)列容量
    private int queueCapacity;
    //保持時(shí)間
    private int keepAliveSeconds;
    //名稱(chēng)前綴
    private String preFix;
    @Bean("MyExecutor")
    public Executor myExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        executor.setThreadNamePrefix(preFix);
        executor.setRejectedExecutionHandler( new ThreadPoolExecutor.AbortPolicy());
        executor.initialize();
        return executor;
    }
}

1.4.3.2 使用自定義線(xiàn)程池

在方法上的@Async注解處指定線(xiàn)程池名字：

@Component
public class MyAsyncTask {
     @Async("MyExecutor") //使用自定義的線(xiàn)程池(執(zhí)行器)
    public void asyncCpsItemImportTask(Long platformId, String jsonList){
        //...具體業(yè)務(wù)邏輯
    }
}

1.4.4 Spring中的線(xiàn)程池(執(zhí)行器)

Spring用TaskExecutor和TaskScheduler接口提供了異步執(zhí)行和調(diào)度任務(wù)的抽象。

Spring的TaskExecutor和java.util.concurrent.Executor接口時(shí)一樣的，這個(gè)接口只有一個(gè)方法execute(Runnable task)。

Spring已經(jīng)內(nèi)置了許多TaskExecutor的實(shí)現(xiàn)，沒(méi)有必要自己去實(shí)現(xiàn)：

SimpleAsyncTaskExecutor：這種實(shí)現(xiàn)不會(huì)重用任何線(xiàn)程，每次調(diào)用都會(huì)創(chuàng)建一個(gè)新的線(xiàn)程。
SyncTaskExecutor：這種實(shí)現(xiàn)不會(huì)異步的執(zhí)行，相反，每次調(diào)用都在發(fā)起調(diào)用的線(xiàn)程中執(zhí)行。它的主要用處是在不需要多線(xiàn)程的時(shí)候，比如簡(jiǎn)單的測(cè)試用例；
ConcurrentTaskExecutor：這個(gè)實(shí)現(xiàn)是對(duì)Java 5 java.util.concurrent.Executor類(lèi)的包裝。有另一個(gè)ThreadPoolTaskExecutor類(lèi)更為好用，它暴露了Executor的配置參數(shù)作為bean屬性。
點(diǎn)擊了解Spring線(xiàn)程池ThreadPoolTaskExecutor講解
SimpleThreadPoolTaskExecutor：這個(gè)實(shí)現(xiàn)實(shí)際上是Quartz的SimpleThreadPool類(lèi)的子類(lèi)，它會(huì)監(jiān)聽(tīng)Spring的生命周期回調(diào)。當(dāng)有線(xiàn)程池，需要在Quartz和非Quartz組件中共用時(shí)，這是它的典型用處。
ThreadPoolTaskExecutor：這是最常用、最通用的一種實(shí)現(xiàn)。它包含了java.util.concurrent.ThreadPoolExecutor的屬性，并且用TaskExecutor進(jìn)行包裝。

1.5 異步中的事務(wù)和返回

1.5.1 異步事務(wù)

在@Async標(biāo)注的方法，同時(shí)也使用@Transactional進(jìn)行標(biāo)注；在其調(diào)用數(shù)據(jù)庫(kù)操作之時(shí)，將無(wú)法產(chǎn)生事務(wù)管理的控制，原因就在于其是基于異步處理的操作。那該如何給這些操作添加事務(wù)管理呢？可以將需要事務(wù)管理操作的方法放置到異步方法內(nèi)部，在內(nèi)部被調(diào)用的方法上添加@Transactional

示例：

方法A：使用了@Async/@Transactional來(lái)標(biāo)注，但是無(wú)法產(chǎn)生事務(wù)控制的目的。
方法B：使用了@Async來(lái)標(biāo)注，B中調(diào)用了C、D，C/D分別使用@Transactional做了標(biāo)注，則可實(shí)現(xiàn)事務(wù)控制的目的

1.5.2 異步返回

異步的業(yè)務(wù)邏輯處理場(chǎng)景有兩種：一個(gè)是不需要返回結(jié)果，另一種是需要接收返回結(jié)果。不需要返回結(jié)果的比較簡(jiǎn)單，就不多說(shuō)了。

需要接收返回結(jié)果的示例如下：

@Async("MyExecutor")
public Future<Map<Long, List>> queryMap(List ids) {
    List<> result = businessService.queryMap(ids);
    ..............
    Map<Long, List> resultMap = Maps.newHashMap();
    ...
    return new AsyncResult<>(resultMap);
}

調(diào)用異步方法的示例：

public Map<Long, List> asyncProcess(List<BindDeviceDO> bindDevices,List<BindStaffDO> bindStaffs, String dccId) {
      Map<Long, List> finalMap =null;
      // 返回值：
      Future<Map<Long, List>> asyncResult = MyService.queryMap(ids);
      try {
          finalMap = asyncResult.get();
      } catch (Exception e) {
          ...
      }
      return finalMap;
}

1.6 異步不能回調(diào)問(wèn)題

使用了異步但是執(zhí)行異步的方法，原因是在方法上加了@Async注解，之所以加這個(gè)注解是因?yàn)閳?bào)錯(cuò)：

There was an unexpected error (type=Internal Server Error, status=500).
Async support must be enabled on a servlet and for all filters involved in async request processing. This is done in Java code using the Servlet API or by adding <async-supported>true</async-supported> to servlet and filter declarations in web.xml

異步測(cè)試時(shí)一直報(bào)這個(gè)錯(cuò)誤，提示我在web.xml開(kāi)啟異步支持，但是我是SpringBoot項(xiàng)目，于是開(kāi)始網(wǎng)上查找

錯(cuò)誤：加@Async注解，會(huì)更加異步，不能獲取異步結(jié)果

正確：根本原因是容器注冊(cè)問(wèn)題，在springboot啟動(dòng)類(lèi)的注解@SpringBootApplication旁邊添加了@ServletComponentScan，才導(dǎo)致上面的報(bào)錯(cuò)和不能回調(diào)，

有三種解決方法：

去掉注解@ServletComponentScan
添加容器注冊(cè)(springboot項(xiàng)目)

  @Bean
    public ServletRegistrationBean dispatcherServlet() {
        ServletRegistrationBean registration = new ServletRegistrationBean(
                new DispatcherServlet(), "/");
        registration.setAsyncSupported(true);
        return registration;
    }
    @Bean
    DispatcherServlet dispatcherServlet(){
        return new DispatcherServlet();
    }

在過(guò)濾器那里添加asyncSupported = true的支持

@WebFilter(urlPatterns="/*",asyncSupported = true)

修改web.xml(傳統(tǒng)xml項(xiàng)目)

需要在 web.xml 文件中的 servlet 定義中添加："<async-supported>true</async-supported>"

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns="http://java.sun.com/xml/ns/javaee"
	xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
	version="3.0">
	<display-name>Archetype Created Web Application</display-name>
	<context-param>
		<param-name>contextConfigLocation</param-name>
		<param-value>classpath:spring-mybatis.xml</param-value>
	</context-param>
	<context-param>
		<param-name>spring.profiles.active</param-name>
		<param-value>dev</param-value>
	</context-param>
	<context-param>
		<param-name>spring.profiles.default</param-name>
		<param-value>dev</param-value>
	</context-param>
	<filter>
		<filter-name>encodingFilter</filter-name>
		<filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
		<async-supported>true</async-supported>
		<init-param>
			<param-name>encoding</param-name>
			<param-value>UTF-8</param-value>
		</init-param>
		<init-param>
			<param-name>forceEncoding</param-name>
			<param-value>true</param-value>
		</init-param>
	</filter>
	<filter-mapping>
		<filter-name>encodingFilter</filter-name>
		<url-pattern>/*</url-pattern>
	</filter-mapping>
	<listener>
		<listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
	</listener>
	<servlet>
		<servlet-name>SpringMVC</servlet-name>
		<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
		<init-param>
			<param-name>contextConfigLocation</param-name>
			<param-value>classpath:spring-mvc.xml</param-value>
		</init-param>
		<load-on-startup>1</load-on-startup>
		<async-supported>true</async-supported>
	</servlet>
	<servlet-mapping>
		<servlet-name>SpringMVC</servlet-name>
		<url-pattern>/</url-pattern>
	</servlet-mapping>
</web-app>