在项目中，如何使用线程池？

前言

有关线程的使用，阿里巴巴开发手册有三点强制要求

创建线程或线程池时请指定有意义的线程名称，方便出错时回溯。
线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。
线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

强制第2点的原因

使用线程池的好处是减少在创建和销毁线程上所花的时间以及系统资源的开销，解决资源不足的问题。

如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者 “过度切换”的问题。

强制第3点的原因

Executors 返回的线程池对象的弊端如下：

「FixedThreadPool和SingleThreadPool」:允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。

「CachedThreadPool 和 ScheduledThreadPool」:允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

我们来看下源码:

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
}

上面这两种方式创建线程池使用的阻塞队列是LinkedBlockingQueue,我们再来看下:

/**
  * 2的31次方，然后在减1 2147483647
  */
@Native public static final int   MAX_VALUE = 0x7fffffff;
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

不设大小理论上队列容量无上限，所以可能会堆积大量请求从而导致OOM。所以在实际开发中我们不建议用Executors来创建线程池。

项目中如何用

在项目中，我们通常有两种方式创建线程池：

第一种：静态方式
第二种：使用Spring Boot创建线程池

比如说我们项目中需要处理用户登录日志，但是此时不想因为记录登录日志耽搁了登录。

如果我们使用同步的方式，可能会因为一些不太需要实时结果的，并且又耗时的业务,可能会导致整个业务变慢：

耗时：200ms=100ms+100ms

如果使用线程池做了异步化后，直接创建个任务丢到线程池里，这样就减少了后面那100ms的等待时间。

在实际项目中，也有很多项目使用消息队列来做异步化，这个看项目情况来，比如：开发成本、后期运维成本等。

静态方式创建线程池

静态方式就是当做一种工具类使用，代码实现如下：

package com.jincou.http.pool;
 
import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
//这里的相关只是一个演示，大家在定参数时，还是要以具体情况来
public class ThreadPoolUtil {
    //获取CPU核数
    static int cpuNums = Runtime.getRuntime().availableProcessors();
    /** 线程池核心池的大小*/
    private static int corePoolSize = 10;
    /** 线程池的最大线程数*/
    private static int maximumPoolSize = cpuNums * 5;
    /** 阻塞队列容量*/
    private static int queueCapacity = 100;
    /** 活跃时间*/
    private static int keepAliveTimeSecond = 300;
 
    public static ExecutorService httpApiThreadPool = null;
 
    static{
        log.info("创建线程数:"+corePoolSize+",最大线程数:"+maximumPoolSize);
        //建立10个核心线程，线程请求个数超过20，则进入队列等待
        httpApiThreadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTimeSecond,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(queueCapacity),new ThreadFactoryBuilder().setNameFormat("PROS-%d").build());
    }
}

在业务代码中的使用：

ThreadPoolUtil.httpApiThreadPool.submit(new Thread(new Runnable() {
                @Override
                public void run() {
                    log.info("======== 登录日志记录=====start=======");
                    try {
                        // TODO: 2022/4/14 业务处理
                        Thread.sleep(1000L);
                        System.out.println("userId=" + userId);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.info("========登录日志记录------end=======");
                }
            }));

很简单吧！但是这种方式存在很多问题，很多项目也是这么在用的。

比如想动态修改线程池参数，这种方式就不好处理了

Spring Boot创建线程池

application.yaml

threadpool:
  corePoolSize: 8
  maxPoolSize: 16
  queueCapacity: 5
  keepAliveSeconds: 300

属性类

@Data
@ConfigurationProperties(prefix = ThreadpoolProperties.PREFIX)
public class ThreadpoolProperties {

    public static final String PREFIX = "threadpool";
    /**核心线程数*/
    private Integer corePoolSize;
    /**最大线程数*/
    private Integer maxPoolSize;
    /**队列存储数*/
    private Integer queueCapacity;
    /**最大线程存活时间*/
    private Integer keepAliveSeconds;
}

配置类

@Configuration
@EnableConfigurationProperties(value = ThreadpoolProperties.class)
@EnableAsync //开启异步请求
public class ThreadPoolConfig {
    
    /**
     * 创建线程池
     */
    @Bean("logExecutor")
    public ThreadPoolTaskExecutor taskExecutor(ThreadpoolProperties properties) {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        pool.setThreadNamePrefix("-----日志线程池-------");
        pool.setCorePoolSize(properties.getCorePoolSize());
        pool.setMaxPoolSize(properties.getMaxPoolSize());
        pool.setKeepAliveSeconds(properties.getKeepAliveSeconds());
        pool.setQueueCapacity(properties.getQueueCapacity());
        // 指定拒绝策略
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 初始化
        pool.initialize();
        return pool;
    }
}

在项目中使用

@Slf4j
@Component
public class AsyncLogTask {

    @Async("logExecutor")
    public void recordLoginLog(Integer userId) {
        log.info("开始处理 登陆日志id = {}", userId);

        // TODO: 假设业务处理

        log.info("处理结束 登陆日志id = {}", userId);
    }
}

controller层

@Slf4j
@RestController
public class TestController {

    @Autowired
   private AsyncLogTask asyncLogTask;

    @RequestMapping(value = "/login")
    public void queryUser( Integer userId) {
        log.info("======正常登陆操作======");
        asyncLogTask.recordLoginLog(userId);
    }
}

输出日志：

2022-04-27 19:56:28.925  INFO 84086 --- [nio-8084-exec-4] c.j.cache.controller.TestController      : ======正常登陆操作======
2022-04-27 19:56:28.930  INFO 84086 --- [--日志线程池-------2] com.jincou.cache.config.AsyncLogTask     : 开始处理 登陆日志id = 1
2022-04-27 19:56:28.930  INFO 84086 --- [--日志线程池-------2] com.jincou.cache.config.AsyncLogTask     : 处理结束 登陆日志id = 1

好了，以上就是我们项目中通常使用的方式，另外，注意，在项目中通常是将注解@EnableAsync 放到项目启动类上。