springboot 等待批量异步任务完成

朱雀 2024-02-18 11:00 5阅读 0赞

以下代码示例的场景是;  
现有需要统计或者计算, 每个计算项不互相依赖, 且最后的计算需要所有异步执行结果的情况下, 使用这种方法.

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;
        }
    
    }