【随笔】Springboot配置线程池

在开发中我们一般都是使用JUC包中的ThreadPoolExecutor的类，但在Springboot项目环境中可以使用ThreadPoolTaskExecutor类完成线程池的声明定义，且还可以使用@Async注解标注在接口实现方法上说明该逻辑异步处理。但是我们在使用的时候务必要进行相应环境配置，否则会存在一些问题，如默认值corePoolSize=1就相当于单线程，queyeCapacity=Integer.MAX_VALUE容易导致OOM问题等。

说明：可以在应用配置直接添加配置属性，我是独立到单独配置文件thread-pool.properties中

#配置核心线程数
thread-pool.corePoolSize = 4

# 配置最大线程数
thread-pool.maxPoolSize = 100

# 配置队列大小
thread-pool.queueCapacity = 500

# 线程最多存活时间/s
thread-pool.keepAliveSeconds = 120

# 配置线程池中的线程的名称前缀
thread-pool.threadNamePrefix = my-thread-

声明一个配置属性对应的javabean,使用@ConfigurationProperties完成配置读取

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


@Data
@ConfigurationProperties(prefix = "thread-pool")
public class MyThreadPoolProperties {

    private int  corePoolSize;

    private int  maxPoolSize;

    private int  queueCapacity;

    private String  threadNamePrefix;

    private int keepAliveSeconds;
}

import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;


@Configuration
@PropertySource(name="thread-pool",value = {"classpath:thread-pool.properties"})
@EnableConfigurationProperties(MyThreadPoolProperties.class)
public class MyThreadPoolConfiguration {

    @Bean("myThreadExecutor")
    public ThreadPoolTaskExecutor myThreadExecutor(MyThreadPoolProperties properties){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getCorePoolSize());
        executor.setMaxPoolSize(properties.getMaxPoolSize());
        executor.setQueueCapacity(properties.getQueueCapacity());
        executor.setThreadNamePrefix(properties.getThreadNamePrefix());

        executor.setKeepAliveSeconds(properties.getKeepAliveSeconds());

        // 拒绝策略,交给调用这线程处理
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        // 初始化
        executor.initialize();

        return executor;
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;


@RestController
@RequestMapping("/api/pools")
public class TestThreadPoolController {

    @Autowired
    @Qualifier("myThreadExecutor")
    private ThreadPoolTaskExecutor executor;

    @GetMapping("/test1")
    public String test1() throws ExecutionException, InterruptedException {

        // 无返回值
        executor.execute(() -> System.out.println("任务被执行了"));

        // 有返回值
        Future result = executor.submit(() -> "同步任务执行成功");
        System.out.println(result.get());

        return "SUCCESS";
    }
}

import com.imk.cases.exceptions.handling.util.AppContextUtils;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;

// 启动异步任务和定时任务
@EnableAsync
@EnableScheduling
@SpringBootApplication
public class CaseExceptionHandlingApplication {

    public static void main(String[] args) {
        ConfigurableApplicationContext applicationContext = SpringApplication.run(CaseExceptionHandlingApplication.class, args);
        AppContextUtils.setApplicationContext(applicationContext);
    }

}
// 定义一个实现方法验证
// 每5s执行一次，使用自定义的线程池
    @Scheduled(cron = "*/5 * * * * ?")
    @Async("myThreadExecutor")
    public void test2(){
        System.out.println("myThreadExecutor test2");
    }

线程池的工作机制原理：

• 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
• 当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
• 如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急。
• 如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。

jdk中提供了 4 种拒绝策略实现：

• AbortPolicy 让调用者抛出 RejectedExecutionException 异常，默认策略
• CallerRunsPolicy 让调用者运行任务
• DiscardPolicy 放弃本次任务
• DiscardOldestPolicy 放弃队列中最早的任务，当前任务取而代之。

关于线程池更多知识请学习JUC中线程池相关知识。