关于spring异步任务

简单地说，用@Async注释bean的方法将使其在单独的线程中执行。换句话说，调用者不会等待被调用方法的完成。利用spring提供的注解即可简单轻松的实现异步任务处理。

默认线程池问题

Spring 异步任务默认使用 Spring 内部线程池 SimpleAsyncTaskExecutor 这个线程池比较坑爹，不会复用线程。也就是说来一个请求，将会新建一个线程。极端情况下，如果调用次数过多，将会创建大量线程。

Java 中的线程是会占用一定的内存空间，所以创建大量的线程将会导致 OOM 错误。
所以如果需要使用异步任务，我们需要一定要使用自定义线程池替换默认线程池。

实战案例

此处以用户注册同时发邮件为例，将发送邮件设置为异步任务。

创建配置类

该配置类中包括了统一异常处理、自定义线程池

@Slf4j
@EnableAsync    // 开启 Spring 异步任务支持
@Configuration
public class AsyncPoolConfig implements AsyncConfigurer {/*** 将自定义的线程池注入到 Spring 容器中
* */@Bean(name = "threadPoolTaskExecutor")@Overridepublic Executor getAsyncExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(20);executor.setQueueCapacity(20);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("My-Async-");   // 这个非常重要// 等待所有任务结果候再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(60);// 定义拒绝策略executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());// 初始化线程池, 初始化 core 线程executor.initialize();return executor;}/*** 指定系统中的异步任务在出现异常时使用到的处理器
* */@Overridepublic AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {return new AsyncExceptionHandler();}/*** 异步任务异常捕获处理器
* */@SuppressWarnings("all")class AsyncExceptionHandler implements AsyncUncaughtExceptionHandler {@Overridepublic void handleUncaughtException(Throwable throwable, Method method,Object... objects) {throwable.printStackTrace();log.error("Async Error: [{}], Method: [{}], Param: [{}]",throwable.getMessage(), method.getName(),JSON.toJSONString(objects));// TODO 发送邮件或者是短信, 做进一步的报警处理}}
}

创建EmailService

@Slf4j
@Service
public class EmailService {// 模拟发送邮件@Async("threadPoolTaskExecutor")public void sendEmail(){try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}log.info("sendEmail on a thread named: [{}]", Thread.currentThread().getName());}// 带返回类型的方法，获取返回值阻塞方式@Async("threadPoolTaskExecutor")public Future sendEmailWithResult(){try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}return new AsyncResult<>("sendEmailWithResult on a thread named: "+Thread.currentThread().getName());}// 带返回类型的方法，获取返回值非阻塞@Async("threadPoolTaskExecutor")public ListenableFuture sendEmailWithAsyncResult(){try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}return new AsyncResult<>("sendEmailWithAsyncResult on a thread named: "+Thread.currentThread().getName());}
}

创建UserController

@Slf4j
@RestController
public class UserController {private EmailService emailService;public UserController(EmailService emailService){this.emailService = emailService;}// 使用异步方式耗时@GetMapping("/register")public String register(){long start = System.currentTimeMillis();emailService.sendEmail();long end = System.currentTimeMillis();return "User register took "+(end -start) +" milliseconds on thread named: "+Thread.currentThread().getName();}// 使用带有返回值的异步方式@GetMapping("/registerWithResult")public String registerWithResult(){long start = System.currentTimeMillis();Future future = emailService.sendEmailWithResult();try {// Future#get 方法将会一直阻塞，直到异步任务执行成功String result = future.get();log.info(result);} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}long end = System.currentTimeMillis();return "registerWithResult on thread named: "+Thread.currentThread().getName();}// 使用带有返回值的异步方式@GetMapping("/registerWithAsyncResult")public String registerWithAsyncResult(){ListenableFuture listenableFuture = emailService.sendEmailWithAsyncResult();// 添加异步回调逻辑listenableFuture.addCallback(new SuccessCallback() {@Overridepublic void onSuccess(String result) {log.info("发送邮件成功，异步回调结果：" + result);}}, new FailureCallback() {@Overridepublic void onFailure(Throwable ex) {log.info("发送邮件失败，异步回调异常：" + ex);}});return "registerWithAsyncResult on thread named: "+Thread.currentThread().getName();}
}