SpringBoot使用@Async异步任务并合并结果集案例

水深无声 2023-07-16 07:59 11阅读 0赞

### 一、基础任务类 ###

@Component
    public class AsyncCheckRuleBaseTask {
    
        @Async
        public Future<Integer> task1() throws InterruptedException {
            System.out.println("task1任务开始");
            long currentTimeMillis = System.currentTimeMillis();
            int sum = 0;
            for (int i = 1; i <= 10000; i++) {
                sum += i;
            }
            Thread.sleep(1000);
            long currentTimeMillis1 = System.currentTimeMillis();
            System.out.println("task1任务耗时:"+(currentTimeMillis1-currentTimeMillis)+"ms");
            return new AsyncResult<>(sum);
        }
    
        @Async
        public Future<Integer> task2() throws InterruptedException {
            System.out.println("task2任务开始");
            long currentTimeMillis = System.currentTimeMillis();
            int sum = 0;
            for (int i = 1; i <= 20000; i++) {
                sum += i;
            }
            Thread.sleep(2000);
            long currentTimeMillis1 = System.currentTimeMillis();
            System.out.println("task2任务耗时:"+(currentTimeMillis1-currentTimeMillis)+"ms");
            return new AsyncResult<>(sum);
        }
    
        @Async
        public Future<Integer> task3() throws InterruptedException {
            System.out.println("task3任务开始");
            long currentTimeMillis = System.currentTimeMillis();
            int sum = 0;
            for (int i = 1; i <= 30000; i++) {
                sum += i;
            }
            Thread.sleep(3000);
            long currentTimeMillis1 = System.currentTimeMillis();
            System.out.println("task3任务耗时:"+(currentTimeMillis1-currentTimeMillis)+"ms");
            return new AsyncResult<>(sum);
        }
    }

### 二、任务调用类  ###

@Service
    public class AsyncCheckRuleTaskService {
    
        @Autowired
        AsyncCheckRuleBaseTask asyncCheckRuleBaseTask;
    
        public List<Integer> doAllTask() throws InterruptedException, ExecutionException {
            long currentTimeMillis = System.currentTimeMillis();
            List<Integer> list=new ArrayList<>();
            Future<Integer> integerFuture1 = asyncCheckRuleBaseTask.task1();
            Future<Integer> integerFuture2 = asyncCheckRuleBaseTask.task2();
            Future<Integer> integerFuture3 = asyncCheckRuleBaseTask.task3();
            //获取异步任务的处理结果，异步任务没有处理完成，会一直阻塞，可以设置超时时间，使用 get 的重载方法
            list.add(integerFuture1.get());
            list.add(integerFuture2.get());
            list.add(integerFuture3.get());
            long currentTimeMillis1 = System.currentTimeMillis();
            System.out.println("task任务总耗时:"+(currentTimeMillis1-currentTimeMillis)+"ms");
            return list;
        }
    }

### 三、测试类  ###

@RestController
    @RequestMapping("/ruleCheckTask")
    public class RuleCheckTaskController {
    
        @Autowired
        AsyncCheckRuleTaskService asyncCheckRuleTaskService;
    
        @PostMapping("/test")
        public ResponseMsg<List<Integer>> check(@RequestBody RequestMsg<ExamineInVo> in) throws InterruptedException, ExecutionException {
            ResponseMsg<List<Integer>> response = new ResponseMsg<>();
            List<Integer> integers = asyncCheckRuleTaskService.doAllTask();
            response.setData(integers);
            response.setHead(ResponseHead.buildSuccessHead());
            return response;
        }
    }

### 四、测试结果 ###

没开异步任务前至少需要：【task任务总耗时:6000ms】

开启以后：【task任务总耗时:3000+ms】

结果如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3htMzkzMzkyNjI1_size_16_color_FFFFFF_t_70][]

### 五、可以自定义异步线程池 ###

public class ContextCopyingDecorator implements TaskDecorator {
        //把context传递到线程中
        @Override
        public Runnable decorate(Runnable runnable) {
            RequestAttributes context = RequestContextHolder.currentRequestAttributes();
            return () -> {
                try {
                    RequestContextHolder.setRequestAttributes(context);
                    runnable.run();
                } finally {
                    RequestContextHolder.resetRequestAttributes();
                }
            };
        }
    }

@Configuration
    @EnableAsync
    public class AsyncThreadPoolConfig {
        /**
         * 获取当前机器CPU数量
         */
        private static final int CPU = Runtime.getRuntime().availableProcessors();
        /**
         * 核心线程数（默认线程数）
         */
        private static final int CORE_POOL_SIZE = CPU;
        /**
         * 最大线程数
         */
        private static final int MAX_POOL_SIZE = CPU * 2;
        /**
         * 允许线程空闲时间（单位：默认为秒）
         */
        private static final int KEEP_ALIVE_TIME = 60;
        /**
         * 缓冲队列数
         */
        private static final int QUEUE_CAPACITY = 200;
        /**
         * 线程池名前缀
         */
        private static final String THREAD_NAME_PREFIX = "ruleTaskExecutor-";
    
        /**
         * 注入自定义ruleTaskExecutor线程池
         * @return
         */
        @Bean("taskExecutor")
        public ThreadPoolTaskExecutor taskExecutor(){
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setTaskDecorator(new ContextCopyingDecorator());
            executor.setCorePoolSize(CORE_POOL_SIZE);
            executor.setMaxPoolSize(MAX_POOL_SIZE);
            executor.setQueueCapacity(QUEUE_CAPACITY);
            executor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
            executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
            // 线程池对拒绝任务的处理策略
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            // 初始化
            executor.initialize();
            return executor;
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3htMzkzMzkyNjI1_size_16_color_FFFFFF_t_70]: /images/20230528/bb1d5650c39b40f6ba89090a42a3aed7.png