以下代码示例的场景是;
现有需要统计或者计算, 每个计算项不互相依赖, 且最后的计算需要所有异步执行结果的情况下, 使用这种方法.
for (int j = 0; j < aaa; j++) {
List<Future<Boolean>> futureList = new ArrayList<>();
// 异步方法1
futureList.add(hgDrugQiHeService.sjStatistics1(recordsList, qiLiHe));
// 异步方法2
futureList.add(hgDrugQiHeService.sjStatistics2(recordsList, qiLiHe));
// 异步方法3
futureList.add(hgDrugQiHeService.sjStatistics3(recordsList, qiLiHe));
//查询任务执行的结果
for (Future<?> future : futureList) {
//CPU高速轮询:每个future都并发轮循,判断完成状态然后获取结果,这一行,是本实现方案的精髓所在。即有10个future在高速轮询,完成一个future的获取结果,就关闭一个轮询
while (true) {
//获取future成功完成状态,如果想要限制每个任务的超时时间,取消本行的状态判断+future.get(1000*1, TimeUnit.MILLISECONDS)+catch超时异常使用即可。
if (future.isDone() && !future.isCancelled()) {
//获取结果
Boolean result = (Boolean) future.get();
System.out.println("任务i=" + i + "获取完成!" + new Date());
//当前future获取结果完毕,跳出while
break;
} else {
//每次轮询休息1毫秒(CPU纳秒级),避免CPU高速轮循耗空CPU---》新手别忘记这个
Thread.sleep(1);
}
System.out.println(new Date());
}
}
// 异步任务全部执行完毕后, 执行以下代码
System.out.println("开始下次循环");
}
@Async("scorePoolTaskExecutor") // 自定义线程池
public Future<Boolean> sjStatistics(){
// 业务代码...
log.info("执行完毕");
return new AsyncResult<>(true);
}
@Configuration
@EnableAsync
public class SyncConfig {
/**
* 为 @Async 自定义一个线程池
*
* @return
*/
@Bean(name = "scorePoolTaskExecutor")
public ThreadPoolTaskExecutor getScorePoolTaskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
int i = Runtime.getRuntime().availableProcessors();
System.out.println("核心线程数"+i);
// 核心线程数
taskExecutor.setCorePoolSize(i*2);
// 线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程池的线程
taskExecutor.setMaxPoolSize(i*2+1);
// 缓冲队列
taskExecutor.setQueueCapacity(200);
// 允许线程的空闲时间60秒:当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
taskExecutor.setKeepAliveSeconds(60);
// 线程池名的前缀
taskExecutor.setThreadNamePrefix("score-");
// 线程池对拒绝任务的处理策略
/**
* 当线程池的任务缓冲队列已满并且线程池中的线程数目已达到MaxPoolSize,如果还有任务到来就会采取任务拒绝策略
* 通常有以下四种策略:
* new ThreadPoolExecutor.AbortPolicy(); 默认,丢弃任务,抛出异常
* new ThreadPoolExecutor.DiscardPolicy(); 丢弃任务,不抛异常
* new ThreadPoolExecutor.DiscardOldestPolicy(); 丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
* new ThreadPoolExecutor.CallerRunsPolicy(); 重试添加当前的任务,自动重复调用 execute() 方法, 直到成功
*/
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();
return taskExecutor;
}
}