ThreadPoolExecutor线程池实战

关忆北.

发布于 2021-12-07 16:43:17

4630

发布于 2021-12-07 16:43:17

文章被收录于专栏：关忆北.关忆北.关忆北.

ThreadPoolExecutor线程池实战

Demo已开源至Github，threadDemo

1.配置TreadPoolProperty

ThreadPoolTaskExecutor构造方法：

ThreadPoolExecutor(int corePoolSize,
                        int maximumPoolSize,
                        long keepAliveTime,
                        TimeUnit unit,
                        BlockingQueue<Runnable> workQueue,
                        ThreadFactory threadFactory,
                        RejectedExecutionHandler handler)

为方便全局使用自定义线程池ThreadPoolTaskExecutor，我们建议使用将其初始化所需参数写入到yml/application配置文件中，以达到不入侵源码且改动线程池运行参数的目的。本文将使用 @ConfigurationProperties注解读取yml配置的方式初始化线程池构造方法。如对获取yml有疑问，可以查看我的另一篇文章，专门讲述了如何获取yml里配置的数据。当然如果您用做demo个人练习，也可以使用赋值的方式完成。

文章链接：如何获取yml里的配置数据？

构造方法参数

corePoolSize:核心线程数，一般取CPU物理核心数，线程默认一直存活
maximumPoolSize：最大容纳线程数，一般取核心线程数的2倍
keepAliveTime：非核心线程的闲置超时时间
unit：keepAliveTime闲置时间单位，一般取秒
queueCapacity：工作队列长度
awaitTerminationSeconds：设置此执行程序应该在关闭时阻止的最大秒数
threadNamePrefix：线程前缀名
RejectedExecutionHandler：线程池内阻塞队列已满，无法接受任务的拒绝策略

SpringThreadPoolPropertyConfig.class配置类

@ConfigurationProperties(prefix = "threadpool.property")
public class SpringThreadPoolPropertyConfig {

    private Integer corePoolSize;

    private Integer maxPoolSize;

    private Integer keepAliveSeconds;

    private Integer queueCapacity;

    private Integer awaitTerminationSeconds;

    private String threadNamePrefix;

    private Boolean waitForTasksToCompleteOnShutdown;
}

application.yml

threadpool:
  property:
    corePoolSize: 6
    maxPoolSize: 12
    keepAliveSeconds: 10
    queueCapacity: 1000
    awaitTerminationSeconds: 0
    threadNamePrefix: pool_
    waitForTasksToCompleteOnShutdown: true

SpringThreadPoolConfig.class线程池配置类

@Configuration
@EnableAsync
public class SpringThreadPoolConfig {

    @Autowired
    private SpringThreadPoolPropertyConfig threadPoolConfig;

    @Bean("taskExecutor")
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor customizeThreadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        customizeThreadPoolTaskExecutor.setCorePoolSize(threadPoolConfig.getCorePoolSize());
        customizeThreadPoolTaskExecutor.setMaxPoolSize(threadPoolConfig.getMaxPoolSize());
        customizeThreadPoolTaskExecutor.setKeepAliveSeconds(threadPoolConfig.getKeepAliveSeconds());
        customizeThreadPoolTaskExecutor.setQueueCapacity(threadPoolConfig.getQueueCapacity());
        customizeThreadPoolTaskExecutor.setAwaitTerminationSeconds(threadPoolConfig.getAwaitTerminationSeconds());
        customizeThreadPoolTaskExecutor.setThreadNamePrefix(threadPoolConfig.getThreadNamePrefix());
        customizeThreadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(threadPoolConfig.getWaitForTasksToCompleteOnShutdown());
        customizeThreadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return customizeThreadPoolTaskExecutor;
    }
}

配置完成

使用线程池

@Resource(name = "taskExecutor")

@Slf4j
@Service
public class AsyncServiceImpl implements AsyncService {

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor taskExecutor;

    @Override
    public void processingTimeOperation() {

        /**
         *
         * 在此处理耗时操作业务逻辑，
         * 如数据量大的插入，只需要在业务层将List等段截取后的List在for循环中传入每一个List调用异步线程即可
         */
        try {
            log.info("进入异步");
            int activeCount = taskExecutor.getActiveCount();
            taskExecutor.submit(() -> log.info(String.valueOf(activeCount))).get();
            log.debug("当前线程信息：-{}", taskExecutor.getThreadNamePrefix() + taskExecutor.getActiveCount());
            log.info("异步体执行完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

taskExecutor.submit()中执行异步耗时操作，如大批量插入操作，submit()方法可以返回线程执行结果，无返回值的需求可以使用execute()方法。