SpringBoot异步方法-@Async
2021-01-23 本文已影响0人
小三哥_e3f4
1、我们为什么要使用异步方法
异步调用通常用在发短信、发送邮件、消息推送 、运维凌晨自动化操作等，这些场景实时性要求不高，大多都是推广统计等服务。
我们采用异步的方式来处理这样耗时 或 实时性要求不高的请求，工作线程可以让后台线程来接手，自己可以及时地被释放到线程池中用于进行后续请求的处理，从而提高了整个服务器的吞吐能力。
springboot中如何实现异步方法
步骤一： SpringApplication启动类上添加@EnableAsync注解
@EnableAsync
@SpringBootApplication
public class SystemApplication{
public static void main(String[] args) {
SpringApplication.run(MicStadiumSystemApplication.class, args);
}
}
步骤二： 配置线程池
注：
- 如果是少量异步任务可不进行此步骤配置，保持springboot默认配置即可。
- 如果是批量异步任务需要进行线程池自定义配置。
/**
* 线程池配置
*/
@Configuration
@Slf4j
public class AsyncConfig implements AsyncConfigurer {
// 核心线程数为服务器的cpu核心数
private static int corePoolSize = Runtime.getRuntime().availableProcessors();
// 线程池中允许的最大线程数
private static int maxPoolSize = 2 * corePoolSize + 1;
// 工作队列大小
private static int queueCapacity = 5000;
@Override
public Executor getAsyncExecutor() {
log.info("核心线程数: "+ corePoolSize + " 最大线程数: " + maxPoolSize);
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
//核心线程数
executor.setCorePoolSize(corePoolSize);
//最大线程数
executor.setMaxPoolSize(maxPoolSize);
//队列大小
executor.setQueueCapacity(queueCapacity);
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setThreadNamePrefix("QueryHealthKit-");
// 拒绝策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
ThreadPoolTaskExecutor属性解释。
- poolSize： 线程池中当前线程的数量。
- corePoolSize: 核心线程数。
- maxPoolSize: 最大线程数。
- queueCapacity：任务队列容量。
1、poolSize<corePoolSize：直接创建新的线程处理新来的任务。
2、poolSize>=corePoolSize && 任务队列未满时：新来的任务加入任务队列。
3、poolSize>corePoolSize && 任务队列满了时:
poolSize<maxPoolSize创建新的线程处理新来的任务。
poolSize=maxPoolSize执行拒绝策略。
- waitForTasksToCompleteOnShutdown：关闭时是否等待未完成的任务执行完毕，默认是false。
- allowCoreThreadTimeout：默认情况下核心线程不会退出，可通过将该参数设置为true，让核心线程也退出。
- keepAliveTime: 设置线程可以保持空闲的时间，超过将被终止，默认为60s。
- threadNamePrefix：设置线程前缀，便于观察线程执行情况。
- rejectedExecutionHandler：拒绝策略，使用界队列的时候，如果队列满了，任务添加到线程池的时候就会有问题，此时就需要拒绝策略：
1、AbortPolicy：该策略是线程池的默认策略。丢掉新来的任务并且抛出RejectedExecutionException异常。
2、DiscardPolicy：如果线程池队列满了，直接丢弃任务不会报错。
3、DiscardOldestPolicy：丢弃队列头部任务，也就是丢弃最早的进入队列的任务，然后尝试加入新的任务进入队列。
4、CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务。
5、自定义
步骤三： 创建异步方法并添加@Async注解
@Service
public class AsyncQueryTask {
// 用与返回接口
Map<String, Object> map = new HashMap<>();
/**
* 异步方法
*/
@Async
public Future<Object> asyncQuery() {
// 异步方法
......
return new AsyncResult<>(map);
}
}
注意事项：以下内容需要严格遵守
- 异步不能方法使用static修饰
- 异步类需要使用@Component注解（或其他注解）
- 异步方法不能与它的调用方法在同一个类中
- 在Async方法上标注@Transactional是没用的。 在Async 方法调用的方法上标注@Transactional 有效。
- 异步方法返回值必须是void或Future。
步骤四： 创建调用者方法
如果Async方法的返回值是void，则调用者方法与普通方法无异。
如果Async方法的返回值是Future，以下代码展示对结果的收集。
@Override
public void healthKitJob() {
Date startDate = new Date();
log.info("任务开始时间{}", startDate);
// 保存异步任务列表
List<Future<Object> futureList = Collections.synchronizedList(Lists.newArrayList());
// ......
for (int i = 0; j < 1000; i++) {
// 1.批量执行异步操作
Future<Object> asyncResult = asyncQueryTask.asyncQuery();
// 2.加入任务列表
futureList.add(asyncResult);
}
// 3.结果归集，用迭代器遍历futureList,高速轮询
while (futureList.size() > 0) {
Iterator<Future<Object> ite = futureList.iterator();
//遍历一遍
while (ite.hasNext()) {
Future<Object> ft = ite.next();
//如果任务完成取结果，否则判断下一个任务是否完成
if (ft.isDone()) { // 判断是否执行完毕
try {
Object ftObject = ft.get(); // 获取结果
// 收集返回结果
// ......
} catch (Exception e) {
e.printStackTrace();
} finally {
ite.remove(); // 任务完成移除任务
}
} else {
try {
// 避免CPU高速运转，这里休息2毫秒，CPU纳秒级
Thread.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
// 4.处理归集后的结果
Date endDate = new Date();
log.info("任务结束时间{},用时{}", endDate, endDate.getTime() - startDate.getTime());
}
注:结果归集要等到所有异步方法都返回后才能往下处理。
执行效果
同时有多个线程在处理任务。
image.png
