Spring boot 使用线程池实现无返回的异步处理和有返回的异步处理

爱被打了一巴掌 2023-07-21 11:06 57阅读 0赞

## 1、场景 ##

http请求，业务异步处理；

1）无需返回的，比如发送短信，push消息，邮件发送等附属业务，异步处理，减少http请求总耗时，提升客户体验。

2）需要返回的，前端等待业务处理结果数据，异步处理返交给http请求的主线程，返回前端；如果是单个业务此处可以优化，请查看此博文前一篇和后一篇；如果是多个业务处理，可以使用多个有返回的异步处理，总重汇总结果并返回。

## 2、知识点 ##

1）线程池ThreadPoolTaskExecutor

2）注解：@EnableAsync，@Async("asyncServiceExecutor")

3）返回值封装Future<T>,new AsyncResult<>(T)

## 3、代码示例 ##

### 1）controller ###

package com.liuxd.controller;
    
    import com.liuxd.entity.Responses;
    import com.liuxd.service.TaskService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.Future;
    
    @Slf4j
    @RestController
    public class AsyncCtr {
    
        @Autowired
        private TaskService taskService;
    
        @RequestMapping(value = "/get", method = RequestMethod.GET)
        public Responses<String> getResult() throws Exception {
    
            log.info("收到HTTP请求...");
            long startTime = System.currentTimeMillis();
    
            //1.异步执行_处理数据_无需等待
            taskService.handleData();
    
            //2.异步执行_处理数据_等待处理结果
            Future<String> future = taskService.getData();
            String result = future.get();
    
            log.info("接收HTTP请求线程任务已完成,退出!");
            long endTime = System.currentTimeMillis();
            log.info("http请求总耗时： " + (endTime - startTime) + "ms");
    
            return new Responses<>(0, result, "SUCCESS");
    
        }
    
    }

### 2）service ###

package com.liuxd.service;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.scheduling.annotation.AsyncResult;
    import org.springframework.stereotype.Service;
    
    import java.util.concurrent.Future;
    
    @Slf4j
    @Service
    public class TaskService {
    
        @Async("asyncServiceExecutor")
        public void handleData() {
    
            log.info("调用service无返回异步方法，开始执行...");
            long startTime = System.currentTimeMillis();
            try {
                Thread.sleep(2500L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            log.info("调用service无返回异步方法，执行结束！！");
            long endTime = System.currentTimeMillis();
            log.info("调用service无返回异步方法总耗时： " + (endTime - startTime) + "ms");
    
        }
    
        @Async("asyncServiceExecutor")
        public Future<String> getData(){
    
            log.info("调用service有返回异步方法，开始执行...");
            long startTime = System.currentTimeMillis();
    
            try {
                Thread.sleep(2500L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            log.info("调用service有返回异步方法，执行结束！！");
    
            long endTime = System.currentTimeMillis();
            log.info("调用service有返回异步方法总耗时： " + (endTime - startTime) + "ms");
    
            return new AsyncResult<>("异步处理完成！");
    
        }
    
    
    }

### 3）线程池 ###

package com.liuxd.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * @program: unif-insure-service
     * @description: 线程池
     **/
    @Configuration
    @EnableAsync
    public class ExecutorConfig {
        @Bean
        public Executor asyncServiceExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //配置核心线程数
            executor.setCorePoolSize(5);
            //配置最大线程数
            executor.setMaxPoolSize(5);
            //配置队列大小
            executor.setQueueCapacity(1000);
            //配置线程池中的线程的名称前缀
            executor.setThreadNamePrefix("async-system-");
    
            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //执行初始化
            executor.initialize();
            return executor;
        }
    }

### 4）Responses ###

package com.liuxd.entity;
    
    import lombok.AllArgsConstructor;
    import lombok.Builder;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    
    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public class Responses<T> {
    
        private Integer code;
    
        private String msg;
    
        private T data;
    
    }

### 5）打印结果 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70][]

### 6）结果分析 ###

1) 共三个线程，http请求主线程，异步无返回线程，异步有返回线程

2）主线程不需要等待无返回线程，会等待有返回线程结果，并最终返回

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70]: /images/20230528/5c6bccfb97c04e33825c186cf18db423.png