Spring Boot异步调用@Async
在实际开发中,有时候为了及时处理请求和进行响应,我们可能会多任务同时执行,或者先处理主任务,也就是异步调用,异步调用的实现有很多,例如多线程、定时任务、消息队列等,
一、普通串行执行演示
1.1任务类
假设有三个任务需要处理我们在平时开发中,会按照逻辑顺序依次编写代码;
@Component
public class Task {
public static Random random =new Random();
public void doTaskOne() throws Exception {
System.out.println("开始做任务一");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务一,耗时:" + (end - start) + "毫秒");
}
public void doTaskTwo() throws Exception {
System.out.println("开始做任务二");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(5000));
long end = System.currentTimeMillis();
System.out.println("完成任务二,耗时:" + (end - start) + "毫秒");
}
public void doTaskThree() throws Exception {
System.out.println("开始做任务三");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务三,耗时:" + (end - start) + "毫秒");
}
}
1.2 测试类
@SpringBootTest
class ApplicationTests {
@Test
void contextLoads() {
}
@Autowired
private Task task;
@Test
public void test() throws Exception {
task.doTaskOne();
task.doTaskTwo();
task.doTaskThree();
}
}
1.3运行结果及分析
观察结果发现,先调用的任务一,任务一执行完成才能开始任务二,任务二完成才能执行任务三;这种串行话的执行程序在实际开发中并没以后任务问题
假如有这样一个功能,任务二是发送邮件,任务三是发送短信,实际开发中这两个任务互补干扰;谁先谁后都都可以,如果是串行操作是不是有点效率低了!最好的办法可以让这两个任务并行操作
实现异步的方式:
- 1、使用多线程,在主线程中分别给任务一,任务二另起一个线程来执行任务
- 2、使用@Async注解实现异步
二、@Async使用演示
2.1 在启动类上添加@EnableAsync开启异步
@Async
是Spring内置注解
,用来处理异步任务,在SpringBoot
中同样适用,且在SpringBoot
项目中,除了boot本身的starter外,不需要额外引入依赖。
而要使用@Async
,需要在 启动类上加上@EnableAsync
主动声明来开启异步方法。
@SpringBootApplication
@EnableAsync
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
2.2 编写任务类,在任务方法添加@Async
@Component
public class TaskAsync {
public static Random random = new Random();
@Async
public void doTaskOne() throws Exception {
System.out.println("开始做任务一");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务一,耗时:" + (end - start) + "毫秒");
}
@Async
public void doTaskTwo() throws Exception {
System.out.println("开始做任务二");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务二,耗时:" + (end - start) + "毫秒");
}
@Async
public void doTaskThree() throws Exception {
System.out.println("开始做任务三");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务三,耗时:" + (end - start) + "毫秒");
}
}
2.3测试类
@SpringBootTest
class ApplicationTests {
@Test
void contextLoads() {
}
@Autowired
private Task task;
@Autowired
private TaskAsync taskAsync;
@Test
public void test() throws Exception {
task.doTaskOne();
task.doTaskTwo();
task.doTaskThree();
}
@Test
public void taskAsynctest() throws Exception {
try {
long start = System.currentTimeMillis();
taskAsync.doTaskOne();
taskAsync.doTaskTwo();
taskAsync.doTaskThree();
Thread.sleep(10000);
long end = System.currentTimeMillis();
System.out.println("end = " + (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2.3 多次运行观察结果
然后我们在运行测试类,这个时候输出可能就五花八门了,任意任务都可能先执行完成,也有可能有的方法因为主程序关闭而没有输出。
2.4总结
使用 @Async可以实现程序的异步执行,完成程序优化;但实际使用中需要注意异步失效问题:
被调用方法和调用方法处理同一个类中会导致异步失效
- 同一个类中。 失效的代码
class TestService {
void a() {
this.b();
}
@Async
void b(){}
}
- 正常的代码
class TestService {
void a(){
BService.b();
}
}
class BService() {
@Async
void b(){}
}
从@Async案例找到Spring框架的bug:exposeProxy=true不生效原因大剖析+最佳解决方案
三、@Async + 线程池使用
@Async 异步方法默认使用 Spring 创建 ThreadPoolTaskExecutor (参考TaskExecutionAutoConfiguration),其中默认核心线程数为 8,默认最大队列和默认最大线程数都是 Integer.MAX_VALUE。创建新线程的条件是队列填满时,而这样的配置队列永远不会填满,如果有 @Async 注解标注的方法长期占用线程 (比如 HTTP 长连接等待获取结果),在核心 8 个线程数占用满了之后,新的调用就会进入队列,外部表现为没有执行。
我们可以自定义一个线程池,线程数的设定需要考虑一下要执行的任务是 IO 密集型任务,还是 CPU 密集型任务。对于 CPU 密集型任务,如 CPU 核数 + 1;对于 IO 密集型任务,由于 IO 密集型任务线程并不是一直在执行任务,则应配置尽可能多的线程,如 CPU 核数 * 2。
3.1线程池配置类
接下来给出一个 IO 密集型任务的线程池配置代码
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@Configuration
public class ThreadPoolConfig {
/**
* 核心线程数
*/
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
/**
* 最大线程数
*/
private static final int MAX_POOL_SIZE = CORE_POOL_SIZE * 4 < 256 ? 256 : CORE_POOL_SIZE * 4;
/**
* 允许线程空闲时间(单位为秒)
*/
private static final int KEEP_ALIVE_TIME = 10;
/**
* 缓冲队列数
*/
private static final int QUEUE_CAPACITY = 200;
/**
* 线程池中任务的等待时间,如果超过这个时候还没有销毁就强制销毁
*/
private static final int AWAIT_TERMINATION = 60;
/**
* 用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
*/
private static final Boolean WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN = true;
/**
* 线程池名前缀
*/
private static final String THREAD_NAME_PREFIX = "Spider-ThreadPool-";
@Bean("spiderTaskExecutor")
public ThreadPoolTaskExecutor spiderTaskExecutor () {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
taskExecutor.setThreadNamePrefix(THREAD_NAME_PREFIX);
taskExecutor.setWaitForTasksToCompleteOnShutdown(WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN);
taskExecutor.setAwaitTerminationSeconds(AWAIT_TERMINATION);
/**
* 拒绝策略 => 当pool已经达到max size的时候,如何处理新任务
* CALLER_RUNS:不在新线程中执行任务,而是由调用者所在的线程来执行
* AbortPolicy:直接抛出异常,这是默认策略;
* DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
* DiscardPolicy:直接丢弃任务;
*/
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();
return taskExecutor;
}
}
3.2 修改任务类
在任务类中的方法上添加@Async("spiderTaskExecutor")
@Component
public class TaskAsync {
public static Random random = new Random();
@Async("spiderTaskExecutor")
public void doTaskOne() throws Exception {
System.out.println("开始做任务一");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务一,耗时:" + (end - start) + "毫秒");
}
@Async("spiderTaskExecutor")
public void doTaskTwo() throws Exception {
System.out.println("开始做任务二");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务二,耗时:" + (end - start) + "毫秒");
}
@Async("spiderTaskExecutor")
public void doTaskThree() throws Exception {
System.out.println("开始做任务三");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任务三,耗时:" + (end - start) + "毫秒");
}
}
最后补充一些知识,要合理的控制线程数(比如采集订单信息的同时要采集订单详情和文章信息,订单详情和文章信息可以合并在一个线程中处理),不要滥用。需要考虑什么时候使用 MQ,什么时候开启线程异步处理。推荐一个分析 jstack 文件的工具,IBM Thread and Monitor Dump Analyzer for Java,分析一下正在运行、发生死锁、等待、阻塞的线程。
四 获取异步执行结果
上面演示了@Async,但是有时候除了需要任务并发调度外,我们还需要获取任务的返回值,且在多任务都执行完成后再结束主任务,这个时候又该怎么处理呢?
CompletableFuture使用详解
五源码地址
源码传送带