Spring Boot @Async 異步任務(wù)執(zhí)行方法

1、任務(wù)執(zhí)行和調(diào)度

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

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

1.1、TaskExecutor類型

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

1. SimpleAsyncTaskExecutor  這種實現(xiàn)不會重用任何線程，每次調(diào)用都會創(chuàng)建一個新的線程。
2. SyncTaskExecutor  這種實現(xiàn)不會異步的執(zhí)行
3. ConcurrentTaskExecutor  這種實現(xiàn)是java.util.concurrent.Executor的一個adapter。
4. SimpleThreadPoolTaskExecutor  這種實現(xiàn)實際上是Quartz的SimpleThreadPool的一個子類，它監(jiān)聽Spring的聲明周期回調(diào)。
5. ThreadPoolTaskExecutor  這是最常用最通用的一種實現(xiàn)。它包含了java.util.concurrent.ThreadPoolExecutor的屬性，并且用TaskExecutor進行包裝。

1.2、注解支持調(diào)度和異步執(zhí)行

To enable support for @Scheduled and @Async annotations add @EnableScheduling and @EnableAsync to one of your

@Configuration classes:

@Configuration
@EnableAsync
@EnableScheduling
public class AppConfig {
}

特別注意

The default advice mode for processing @Async annotations is "proxy" which allows for interception of calls through the proxy only; local calls within the same class cannot get intercepted that way. For a more advanced mode of interception, consider switching to "aspectj" mode in combination with compile-time or load-time weaving.

默認是用代理去處理@Async的，因此，相同類中的方法調(diào)用帶@Async的方法是無法異步的，這種情況仍然是同步。

舉個例子：下面這種，在外部直接調(diào)用sayHi()是可以異步執(zhí)行的，而調(diào)用sayHello()時sayHi()仍然是同步執(zhí)行

public class A {
   public void sayHello() {
    sayHi();
  }

  @Async
  public void sayHi() {

  }   
}

1.3、@Async注解

在方法上加@Async注解表示這是一個異步調(diào)用。換句話說，方法的調(diào)用者會立即得到返回，并且實際的方法執(zhí)行是想Spring的TaskExecutor提交了一個任務(wù)。

In other words, the caller will return immediately upon invocation and the actual execution of the method will occur in a task that has been submitted to a Spring TaskExecutor.

@Async
void doSomething() {
  // this will be executed asynchronously
}

@Async
void doSomething(String s) {
  // this will be executed asynchronously
}

@Async
Future<String> returnSomething(int i) {
  // this will be executed asynchronously
}

注意：

@Async methods may not only declare a regular java.util.concurrent.Future return type but also Spring's org.springframework.util.concurrent.ListenableFuture or, as of Spring 4.2, JDK 8's java.util.concurrent.CompletableFuture: for richer interaction with the asynchronous task and for immediate composition with further processing steps.

1.4、@Async限定Executor

默認情況下，當在方法上加@Async注解時，將會使用一個支持注解驅(qū)動的Executor。然而，@Async注解的value值可以指定一個別的Executor

@Async("otherExecutor")
void doSomething(String s) {
  // this will be executed asynchronously by "otherExecutor"
}

這里，otherExecutor是Spring容器中任意Executor bean的名字。

1.5、@Async異常管理

當一個@Async方法有一個Future類型的返回值時，就很容易管理在調(diào)Future的get()方法獲取任務(wù)的執(zhí)行結(jié)果時拋出的異常。如果返回類型是void，那么異常是不會被捕獲到的。

public class MyAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {

  @Override
  public void handleUncaughtException(Throwable ex, Method method, Object... params) {
    // handle exception
  }
}

2、線程池配置

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@Configuration
@EnableAsync
public class TaskExecutorConfig {
  private Integer corePoolSize = 30;
  private Integer maxPoolSize = 50;
  private Integer keepAliveSeconds = 300;
//  private Integer queueCapacity = 2000;
  @Bean("myThreadPoolTaskExecutor")
  public ThreadPoolTaskExecutor myThreadPoolTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setKeepAliveSeconds(keepAliveSeconds);
//    executor.setQueueCapacity(queueCapacity);
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.initialize();
    return executor;
  }
}

調(diào)用

@Async("myThreadPoolTaskExecutor")
  @Override
  public void present(CouponPresentLogEntity entity) {
    try {
      CouponBaseResponse rst = couponSendRpcService.send(entity.getUserId(), entity.getCouponBatchKey(), "1", entity.getVendorId());
      if (null != rst && rst.isSuccess()) {
        entity.setStatus(PresentStatusEnum.SUCCESS.getType());
      }else {
        String reason = (null == rst) ? "響應(yīng)異常" : rst.getMsg();
        entity.setFailureReason(reason);
        entity.setStatus(PresentStatusEnum.FAILURE.getType());
      }
    }catch (Exception ex) {
      log.error(ex.getMessage(), ex);
      entity.setFailureReason(ex.getMessage());
      entity.setStatus(PresentStatusEnum.FAILURE.getType());
    }
    couponPresentLogDao.update(entity);
  }

結(jié)果

[INFO ] 2018-05-09 16:27:39.887 [myThreadPoolTaskExecutor-1] [com.ourhours.coupon.rpc.dubbo.ReceiveLogFilter] - receive method-name:send; arguments:[10046031,"4d7cc32f8f7e4b00bca56f6bf4b3b658","1",10001]
[INFO ] 2018-05-09 16:27:39.889 [myThreadPoolTaskExecutor-2] [com.ourhours.coupon.rpc.dubbo.ReceiveLogFilter] - receive method-name:send; arguments:[10046031,"4d7cc32f8f7e4b00bca56f6bf4b3b658","1",10001]

參考：

Spring Framework Reference Documentation 4.3.17.RELEASE

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。