springboot项目中多线程应用，异步处理，异步调用

springboot项目中多线程应用，异步处理，异步调用

多线程的应用

线程池的应用

异步处理

异步调用

随着开发经验的积累，我们逐渐都了解到了项目中需要多线程的应用或者线程池的应用，有一些耗时的业务需要我们去进行异步调用，特别是针对于一些其他部门所提供的接口，因此码下这篇博客，希望对大家有用！

配置一个线程池

@EnableAsync
@Configuration
@Slf4j
public class ThreadPoolConfig {

    @Bean
    public RestTemplate restTemplate(){
        return new RestTemplate();
    }

    // 当池子大小小于corePoolSize，就新建线程，并处理请求
    // 当池子大小等于corePoolSize，把请求放入workQueue(QueueCapacity)中，池子里的空闲线程就去workQueue中取任务并处理
    // 当workQueue放不下任务时，就新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
    // 当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁
    @Bean(name = "asyncServiceExecutor")
    public Executor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(10);
        executor.setKeepAliveSeconds(30);
        executor.setThreadNamePrefix("async-");
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

}

定义两个额外的比较耗时的测试接口

@Slf4j
@RestController
@RequestMapping(value = "/test")
public class TestController {
  
    @GetMapping(value = "/hello1")
    public JSONObject hello1 (String name) throws InterruptedException {
        log.info("this is hello1 ---> sleep(5000) begin " + new Date());
        Thread.sleep(5000);
        log.info("this is hello1 ---> sleep(5000) end   " + new Date());
        JSONObject res = new JSONObject();
        res.put("name", "hello1-" + name);
        return res;
    }

    @GetMapping(value = "/hello2")
    public JSONObject hello2 (String name) throws InterruptedException {
        log.info("this is hello2 ---> sleep(8000) begin " + new Date());
        Thread.sleep(8000);
        log.info("this is hello2 ---> sleep(8000) end   " + new Date());
        JSONObject res = new JSONObject();
        res.put("name", "hello2-" + name);
        return res;
    }
  
}

定义多线程调用的业务类

@Service
@Slf4j
public class OtherService {

    @Autowired
    private RestTemplate restTemplate;

    @Async("asyncServiceExecutor")
    public CompletableFuture<JSONObject> invokeHello1(String name){
        String url = "http://localhost:8080/test/hello1?name=" + name;
        log.info("多线程开始执行{}, threadId:{}, threadName:{}", url, Thread.currentThread().getId(), Thread.currentThread().getName());
        ResponseEntity<JSONObject> responseEntity = restTemplate.getForEntity(url, JSONObject.class);
        JSONObject jsonRes = responseEntity.getBody();
        log.info("多线程执行完毕{}, 响应Res:{}, threadId:{}, threadName:{}", url, jsonRes, Thread.currentThread().getId(), Thread.currentThread().getName());
        return CompletableFuture.completedFuture(jsonRes);
    }

    @Async("asyncServiceExecutor")
    public CompletableFuture<JSONObject> invokeHello2(String name){
        String url = "http://localhost:8080/test/hello2?name=" + name;
        log.info("多线程开始执行{}, threadId:{}, threadName:{}", url, Thread.currentThread().getId(), Thread.currentThread().getName());
        ResponseEntity<JSONObject> responseEntity = restTemplate.getForEntity(url, JSONObject.class);
        JSONObject jsonRes = responseEntity.getBody();
        log.info("多线程执行完毕{}, 响应Res:{}, threadId:{}, threadName:{}", url, jsonRes, Thread.currentThread().getId(), Thread.currentThread().getName());
        return CompletableFuture.completedFuture(jsonRes);
    }

}

定义要使用多线程的接口

@Slf4j
@RestController
@RequestMapping(value = "/test")
public class TestController {
  
    @Autowired
    private Demo3Service demo3Service;

    @GetMapping(value = "/testHello")
    public JSONObject testHello (String name) throws ExecutionException, InterruptedException {
        return demo3Service.testHello(name);
    }
  
}

定义要使用多线程的servcie

@Service
@Slf4j
public class Demo3Service {

    @Autowired
    private OtherService otherService;

    public JSONObject testHello (String name) throws ExecutionException, InterruptedException {
        log.info("this is demo3Service.testHello({}), begin -> {}", name, new Date());
        // 此处hello1和hello2均为异步调用
        CompletableFuture<JSONObject> hello1Future = otherService.invokeHello1(name);
        CompletableFuture<JSONObject> hello2Future = otherService.invokeHello2(name);
        // 这里可以处理其他的业务逻辑
        log.info("async invoke doing...");
        // 开始处理hello1和hello2的接口相应
        CompletableFuture.allOf(hello1Future, hello2Future).join();
        JSONObject hello1Res = hello1Future.get();
        JSONObject hello2Res = hello2Future.get();
        log.info("this is demo3Service.testHello({}), end -> {}", name, new Date());
        JSONObject jsonRes = new JSONObject();
        jsonRes.put("hello1", hello1Res.get("name"));
        jsonRes.put("hello2", hello2Res.get("name"));
        return jsonRes;
    }
    
}

最后，让我们一起来测试一下 http://localhost:8080/test/testHello?name=zhangsan这个接口

一起来看一下控制台输出的日志

this is demo3Service.testHello(zhangsan), begin -> Mon Apr 11 22:59:13 CST 2022
async invoke doing...
多线程开始执行http://localhost:8080/test/hello1?name=zhangsan, threadId:43, threadName:async-1
多线程开始执行http://localhost:8080/test/hello2?name=zhangsan, threadId:44, threadName:async-2
this is hello2 ---> sleep(8000) begin Mon Apr 11 22:59:13 CST 2022
this is hello1 ---> sleep(5000) begin Mon Apr 11 22:59:13 CST 2022
this is hello1 ---> sleep(5000) end   Mon Apr 11 22:59:18 CST 2022
多线程执行完毕http://localhost:8080/test/hello1?name=zhangsan, 响应Res:{"name":"hello1-zhangsan"}, threadId:43, threadName:async-1
this is hello2 ---> sleep(8000) end   Mon Apr 11 22:59:21 CST 2022
多线程执行完毕http://localhost:8080/test/hello2?name=zhangsan, 响应Res:{"name":"hello2-zhangsan"}, threadId:44, threadName:async-2
this is demo3Service.testHello(zhangsan), end -> Mon Apr 11 22:59:21 CST 2022

OK，这就是多线程/线程池的一个基本使用了，希望能帮到大家