SpringBoot中线程池使用

最近接手的项目遇到一个这样的问题：我们服务的一个接口需要循环查询另一个服务的接口，这样如果数据量较多的时候，我们服务的接口响应时间可能就会超时，伪代码如下：

List<String> listOfResult = new ArrayList<>();
listOfData.stream().forEach(t -> {
// 调用其他微服务查询数据
String result = queryOtherServer(t);
if (null != result){
listOfResult.add(result);
}
});

这里先不讨论为什么不能让对方服务提供一个批量查询接口，然后通过一次调用获取数据结果，历史原因就是这样，短期只能我们系统先优化。

好吧，头疼医头吧。

这里如果要提高性能，最直接的办法就是并发查询，然后汇总查询结果。

优化步骤：

在项目中增加线程池配置，如下：

/**
 * @description 自定义线程池配置
 * @date 2021/7/3
 */
@Configuration
@EnableAsync
public class MyThreadPoolConfig {

    /**
     * 自定义线程池
     * @return
     */
    @Bean("myThreadPool")
    public Executor threadPoolExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建时候初始化的线程数
        executor.setCorePoolSize(10);
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(20);
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(500);
        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setThreadNamePrefix("do-something-");
        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        executor.initialize();
        return executor;
    }
}

这里需要注意的是，一定要使用@EnableAsync。

在需要异步执行的component方法上添加@Async 注解，如下：

/**
 * @author Charlie
 * @description
 * @date 2021/3/29
 */
@Component
public class TestComponent {

    @Async("myThreadPool")
    public CompletableFuture<String> mockHttpRequest(String requestId){
        Long tId = Thread.currentThread().getId();
        System.out.println("线程" + tId + "开始调用，请求ID为：" + requestId + " " + System.currentTimeMillis());
        try {
             Thread.sleep(new Random().nextInt(10) * 100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("线程" + tId + "调用结束，请求ID为：" + requestId + " " + System.currentTimeMillis());
        return CompletableFuture.completedFuture(tId + ":" + requestId);
    }
}

这里之所以返回CompletableFuture是因为需要接收返回值。

调用的时候

@RequestMapping("/testThreadPool")
public void testThreadPool(){
    List<String> list = new ArrayList<>();
    list.add("a");
    list.add("b");
    list.add("c");
    list.add("d");

    List<String> listOfResult = new ArrayList<>();
    List<CompletableFuture<String>> listOfFuture = new ArrayList<>();
    list.stream().forEach(t -> {
CompletableFuture<String> completableFuture = component.mockHttpRequest(t);
            listOfFuture.add(completableFuture);
        });

// 等待所有调用返回结果
CompletableFuture.allOf(listOfFuture.toArray(new CompletableFuture[listOfFuture.size()])).join();

listOfFuture.forEach(future -> {
            try{
                listOfResult.add(future.get());
            }catch (Exception e){
// 异常处理
            }
        });

System.out.println(gson.toJson(listOfResult));

}