springboot @Async 注解如何实现方法异步
作者:Bug布道师
@Async注解如何实现方法异步
处理大批量数据的时候,效率很慢。所以考虑一下使用多线程。
刚开始自己手写的一套,用了线程池启动固定的线程数进行跑批。但是后来老大考虑到自己手写的风险不好控制,所以使用spring的方法。
这里没有详细介绍,只有简单的demo,只会用,不懂原理:
一、springboot的App类需要的注解
package com.xxx.xxx.xxx; import java.util.concurrent.ThreadPoolExecutor; import org.springframework.boot.web.support.SpringBootServletInitializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** * 类功能说明:服务生产者启动类 * <p> * <strong></strong> * </p> * * @version * @author * @since 1.8 */ @Configuration @EnableAsync public class Application extends SpringBootServletInitializer { @Bean public TaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 设置核心线程数 executor.setCorePoolSize(5); // 设置最大线程数 executor.setMaxPoolSize(60); // 设置队列容量 executor.setQueueCapacity(20); // 设置线程活跃时间(秒) executor.setKeepAliveSeconds(60); // 设置默认线程名称 executor.setThreadNamePrefix("what-"); // 设置拒绝策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 等待所有任务结束后再关闭线程池 executor.setWaitForTasksToCompleteOnShutdown(true); return executor; } }
springboot的App类,很简单,就能使用很多东西。
二、service层的注解
package com.xxx.xxx.service.impl; import java.util.concurrent.Future; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Service; import com.xxx.xxx.service.XXXAsyncService ; @Service public class XXXAsyncServiceImpl implements XXXAsyncService { @Async public Future<Long> rtn1() throws Exception { //do something //有返回值的时候,可以返回string,long之类的。 return new AsyncResult<>(1); } @Async public void rtn2() throws Exception { //do something //这个可以没有返回值. } }
三、调用层
package com.xxx.xxx.controller; import java.util.concurrent.Future; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; import com.xxx.xxx.service.XXXAsyncService; @RestController @RequestMapping(value="/xxx") public class XXXAsyncController { @Autowired private XXXAsyncService xxxAsyncService; /** * 这里调用异步方法 */ @RequestMapping(value = "/xxx") public void dodo() throws Exception { int threads = 10;//十个线程 List<Future<Long>> list = new ArrayList<>(); for(int i = 0;i < threads; i++){ //这里循环调用异步方法。 //如果存在大量数据,可以在这里把数据切片,然后循环调用,分批处理数据。效率杠杠的。 list .add(xxxAsyncService.rtn1()); } long count = 0; for(Future<Long> l : tsfCountList) { //异步调用需要返回值的时候,这里可以把返回值都放入到list集合中,然后可以统一处理。 这里的get()是阻塞的,因为需要所以异步方法返回,在继续执行。 count += l.get(); } System.out.println("调用次数:" + count); } }
这些代码全是手写,记录下来,以后用的时候,省的忘了,查起来麻烦。。
异步注解@Async的使用以及注意事项
第一步开启异步
@Configuration @EnableAsync public class SpringAsyncConfig { ... }
默认情况下,@EnableAsync检测Spring的@Async注释和EJB 3.1 javax. EJB .异步;此选项还可用于检测其他用户定义的注释类型。(也可以在SpringBoot的启动类上直接加@EnableAsync注解)
在 Spring 中,用 @Async 注解指定的方法,该方法被调用时会以异步的方式执行。而如果没有在 @Async 注解中指定线程池,就会使用默认的线程池。默认的线程池为 SimpleAsyncTaskExecutor 。
该线程池不会复用线程,每有一个新任务被提交,该线程池就会创建一个新的线程实例用于执行任务。下面为相关的代码:
protected void doExecute(Runnable task) { Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task)); thread.start(); }
而如果想要指定线程池,可以通过在 @Async 注解中的 value 参数中指定所要使用的线程池的 Bean Name 。另一种方法是是一个实现了 AsyncConfigurer 接口或是继承其默认适配器类 AsyncConfigurerSupport 的配置类,这样 @Async 注解的方法就会使用指定的自定义的线程池。
使用@Async注解的话采用的是springBoot默认的线程池,不过一般我们会自定义线程池(因为比较灵活),配置方式有:
- 使用 xml 文件配置的方式
- 使用Java代码结合@Configuration进行配置(推荐使用)
下面显示配置线程的代码实现
package com.deppon.ptos.load.config; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; import java.util.concurrent.ThreadPoolExecutor; /** * @Description: 异步线程管理 * @Author: LYH * @CreateDate: 2019/6/27 8:54 * @Version: 1.0 * @JDK: 1.8 */ @Configuration @EnableAsync @Slf4j public class ExecutorConfig { @Value("${async.executor.thread.core_pool_size}") private int corePoolSize; @Value("${async.executor.thread.max_pool_size}") private int maxPoolSize; @Value("${async.executor.thread.queue_capacity}") private int queueCapacity; @Value("${async.executor.thread.name.prefix}") private String namePrefix; @Bean(name = "asyncServiceExecutor") public Executor asyncServiceExecutor() { log.info("start asyncServiceExecutor"); ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor(); //配置核心线程数 executor.setCorePoolSize(corePoolSize); //配置最大线程数 executor.setMaxPoolSize(maxPoolSize); //配置队列大小 executor.setQueueCapacity(queueCapacity); //配置线程池中的线程的名称前缀 executor.setThreadNamePrefix(namePrefix); // rejection-policy:当pool已经达到max size的时候,如何处理新任务 // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //执行初始化 executor.initialize(); return executor; } }
package com.deppon.ptos.load.config; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import java.util.concurrent.Callable; import java.util.concurrent.Future; import java.util.concurrent.ThreadPoolExecutor; /** * @Description: 打印异步线程的执行情况 使用Callbale Future 来返回线程的信息 * @Author: 633805 LYH * @CreateDate: 2019/6/27 8:59 * @Version: 1.0 * @JDK: 1.8 */ @Component @Slf4j public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { private void showThreadPoolInfo(String prefix) { ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor(); if (null == threadPoolExecutor) { return; } log.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]", this.getThreadNamePrefix(), prefix, threadPoolExecutor.getTaskCount(), threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size()); } @Override public void execute(Runnable task) { showThreadPoolInfo("1. do execute"); super.execute(task); } @Override public void execute(Runnable task, long startTimeout) { showThreadPoolInfo("2. do execute"); super.execute(task, startTimeout); } @Override public Future<?> submit(Runnable task) { showThreadPoolInfo("1. do submit"); return super.submit(task); } @Override public <T> Future<T> submit(Callable<T> task) { showThreadPoolInfo("2. do submit"); return super.submit(task); } @Override public ListenableFuture<?> submitListenable(Runnable task) { showThreadPoolInfo("1. do submitListenable"); return super.submitListenable(task); } @Override public <T> ListenableFuture<T> submitListenable(Callable<T> task) { showThreadPoolInfo("2. do submitListenable"); return super.submitListenable(task); } }
使用:
@Async("asyncServiceExecutor")
到这一步,异步就算开启了。
下面主要说一说错误的
使用@Async导致异步不成功的情况
如下方式会使@Async失效
- 异步方法使用static修饰
- 异步类没有使用@Component注解(或其他注解)导致spring无法扫描到异步类
- 异步方法不能与被调用的异步方法在同一个类中
- 类中需要使用@Autowired或@Resource等注解自动注入,不能自己手动new对象
- 如果使用SpringBoot框架必须在启动类中增加@EnableAsync注解
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。