SpringBoot项目中自定义线程池与异步调用案例

SpringBoot项目中自定义线程池与异步调用案例

一、自定义线程池

1、配置文件

server:
  port: 9006
# 线程池配置参数
task:
  pool:
    corePoolSize: 10 # 设置核心线程数
    maxPoolSize: 20  # 设置最大线程数
    keepAliveTime: 300 # 设置空闲线程存活时间（秒）
    queueCapacity: 100 # 设置队列容量
    threadNamePrefix: "gblfy-signpolicy-asynnotify-" # 设置线程名称前缀
    awaitTerminationSeconds: 60 #  设置线程池等待终止时间(秒)
spring:
  main:
    allow-bean-definition-overriding: true

2、线程池配置属性对应的实体类映射

package com.panda.threadpool.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;

/**
 * @Description: 线程池配置属性映射实体类
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-28  16:58
 */
@Data
@ConfigurationProperties(prefix = "task.pool")
public class ThreadPoolProperties {

    private Integer corePoolSize;

    private Integer maxPoolSize;

    private Integer keepAliveTime;

    private Integer queueCapacity;

    private Integer awaitTerminationSeconds;

    private String threadNamePrefix;
}

3、开启异步线程支持

package com.panda.threadpool;

import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure;
import com.panda.threadpool.config.ThreadPoolProperties;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.scheduling.annotation.EnableAsync;

/**
 * @Description:
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-28  16:55
 */
// 排除数据源和数据库连接池配置 （此项目中引入依赖后没有用到所以要排除）
@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, DruidDataSourceAutoConfigure.class}) 
@EnableAsync // 开启异步线程支持
@EnableConfigurationProperties({ThreadPoolProperties.class}) // 开启配置属性支持
public class ApplicationThreadPool {
    public static void main(String[]args){
        SpringApplication.run(ApplicationThreadPool.class, args);
    }
}

4、创建自定义线程池配置类

package com.panda.threadpool.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Description: 创建自定义线程池配置类
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-28  17:12
 */
@Configuration
public class AsyncScheduledTaskConfig {

    private static final Logger logger = LoggerFactory.getLogger(AsyncScheduledTaskConfig.class);

    @Resource
    private ThreadPoolProperties poolProperties;


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

    /**
     *  1.当池子大小小于corePoolSize，就新建线程，并处理请求
     *  2.当池子大小等于corePoolSize，把请求放入workQueue(QueueCapacity)中，
     *	  池子里的空闲线程就去workQueue中取任务并处理
     *  3.当workQueue放不下任务时，就新建线程入池，并处理请求，
     *	  如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
     *  4.当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，
     *    如果无请求可处理就自行销毁
     * @return
     */

    @Bean("customAsyncThreadPool") // 自定线程池在容器中的实例名称
    public Executor customAsyncThreadPool() {

        logger.info("start asyncServiceExecutor");

        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        // 线程池最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(poolProperties.getMaxPoolSize());
        // 线程池核心线程数
        threadPoolTaskExecutor.setCorePoolSize(poolProperties.getCorePoolSize());
        // 任务队列的大小
        threadPoolTaskExecutor.setQueueCapacity(poolProperties.getQueueCapacity());
        // 线程池名的前缀
        threadPoolTaskExecutor.setThreadNamePrefix(poolProperties.getThreadNamePrefix());
        // 允许线程池的空闲时间30秒
        threadPoolTaskExecutor.setKeepAliveSeconds(poolProperties.getKeepAliveTime());
        // 设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
        threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        // 设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住
        threadPoolTaskExecutor.setAwaitTerminationSeconds(poolProperties.getAwaitTerminationSeconds());

        /**
         * 拒绝处理策略
         * CallerRunsPolicy()：交由调用方线程运行，比如 main 线程。
         * AbortPolicy()：直接抛出异常。
         * DiscardPolicy()：直接丢弃。
         * DiscardOldestPolicy()：丢弃队列中最老的任务。
         */
        /**
         * 特殊说明：
         * 1. 这里演示环境，拒绝策略咱们采用抛出异常
         * 2.真实业务场景会把缓存队列的大小会设置大一些，
         *   如果，提交的任务数量超过最大线程数量或将任务环缓存到本地、redis、mysql中,保证消息不丢失
         * 3.如果项目比较大的话，异步通知种类很多的话，建议采用MQ做异步通知方案
         */
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        // 线程初始化
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

二、实现使用自定义线程池异步调用第三方接口案例

1、控制层

模拟接口testHello并行调用hello1和hello2接口

package com.panda.threadpool.controller;

import com.alibaba.fastjson.JSONObject;
import com.panda.threadpool.service.ThreadPoolService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.ExecutionException;


/**
 * @Description: 异步调用测试类
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-30  11:58
 */
@RestController
@RequestMapping(value = "/test")
public class TestController {

    @Resource
    private ThreadPoolService threadPoolService;

    private static final Logger logger = LoggerFactory.getLogger(TestController.class);

    /**
     *  模拟远程接口1
     * @param name
     * @param age
     * @return
     * @throws InterruptedException
     */
    @GetMapping(value = "/hello1")
    public JSONObject hello1(String name,Integer age) throws InterruptedException {
        logger.info("this is hello1 ---> sleep(5000) begin " + new Date());
        Thread.sleep(5000);
        logger.info("this is hello1 ---> sleep(5000) end   " + new Date());
        JSONObject res = new JSONObject();
        res.put("name", "hello1-" + name);
        res.put("age","hello1-"+age);
        return res;
    }

    /**
     * 模拟远程接口2
     * @param name
     * @param age
     * @return
     * @throws InterruptedException
     */
    @GetMapping(value = "/hello2")
    public JSONObject hello2(String name,Integer age) throws InterruptedException {
        logger.info("this is hello2 ---> sleep(8000) begin " + new Date());
        Thread.sleep(8000);
        logger.info("this is hello2 ---> sleep(8000) end   " + new Date());
        JSONObject res = new JSONObject();
        res.put("name", "hello2-" + name);
        res.put("age","hello2-" + age);
        return res;
    }

    /**
     *  调用远程接口
     * @param name
     * @param age
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @GetMapping(value = "/testHello")
    public JSONObject testHello(@RequestParam("name") String name,
                                @RequestParam("age") Integer age) throws ExecutionException, InterruptedException {
        return threadPoolService.testHello(name,age);
    }


}

2、定义多线程调用的业务类

package com.panda.threadpool.service.impl;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;

/**
 * @Description: 多线程调用业务类
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-30  12:16
 */
@Service
public class OtherService {

    private static final Logger logger = LoggerFactory.getLogger(OtherService.class);

    @Resource
    private RestTemplate restTemplate;

    /**
     * 调用hello1接口
     * @param name
     * @param age
     * @return
     */
    @Async("customAsyncThreadPool") // 使用自定义线程池
    public CompletableFuture<JSONObject> invokeHello1(String name, Integer age) {
        String url = "http://localhost:9006/test/hello1?name=" + name + "&age="+age;
        logger.info("多线程开始执行{}, threadId:{},threadName:{}",
                url,Thread.currentThread().getId(),Thread.currentThread().getName());
        ResponseEntity<JSONObject> responseEntity = restTemplate.getForEntity(url, JSONObject.class);
        JSONObject jsonRes = responseEntity.getBody();
        logger.info("多线程执行完毕{}, 响应Res:{}, threadId:{}, threadName:{}",
                url, jsonRes, Thread.currentThread().getId(), Thread.currentThread().getName());
        return CompletableFuture.completedFuture(jsonRes);
    }

    /**
     * 调用接口hello2
     * @param name
     * @param age
     * @return
     */
    @Async("customAsyncThreadPool") // 使用自定义线程池
    public CompletableFuture<JSONObject> invokeHello2(String name, Integer age) {
        String url = "http://localhost:9006/test/hello2?name=" +name + "&age="+age;
        logger.info("多线程开始执行{},threadId:{},threadName:{}",url,
                Thread.currentThread().getId(),Thread.currentThread().getName());
        ResponseEntity<JSONObject> responseEntity = restTemplate.getForEntity(url, JSONObject.class);
        JSONObject jsonRes = responseEntity.getBody();
        logger.info("多线程执行完毕{},响应Res:{},threadId:{},threadName:{}",
                url,jsonRes,Thread.currentThread().getId(),Thread.currentThread().getName());
        return CompletableFuture.completedFuture(jsonRes);

    }

}

3、定义使用多线程的接口ThreadPoolService

package com.panda.threadpool.service;

import com.alibaba.fastjson.JSONObject;

import java.util.concurrent.ExecutionException;

/**
 * @Description:
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-28  17:33
 */
public interface ThreadPoolService {
    
    JSONObject testHello(String name,Integer age) throws ExecutionException, InterruptedException;
}

4、定义使用多线程的接口实现类ThreadPoolServiceImpl

package com.panda.threadpool.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.panda.threadpool.service.ThreadPoolService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @Description: 自定义线程池异步调用测试
 * @Created: with IntelliJ IDEA.
 * @Author: luzic
 * @CreateTime: 2022-09-28  22:47
 */
@Service
public class ThreadPoolServiceImpl implements ThreadPoolService {

    private static final Logger logger = LoggerFactory.getLogger(ThreadPoolServiceImpl.class);

    @Resource
    private OtherService otherService;

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

5、接口测试

在浏览器中输入：http://localhost:9006/test/testHello?name=xiaozhang&age=99

以上就是SpringBoot实现自定义线程池，并使用自定义线程池实现异步调用接口