别告诉我你连线程池都不会用~ 一文搞懂线程池

admin • 2021-12-17 20:04 • 程序人生

线程池作用

降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
提高响应速度：任务到达时，无需等待线程创建即可立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行

线程池5种状态

RUNNING：线程池一旦被创建，就处于 RUNNING 状态，任务数为 0，能够接收新任务，对已排队的任务进行处理。
SHUTDOWN：不接收新任务，但能处理已排队的任务。调用线程池的 shutdown() 方法，线程池由 RUNNING 转变为 SHUTDOWN 状态。
STOP：不接收新任务，不处理已排队的任务，并且会中断正在处理的任务。调用线程池的 shutdownNow() 方法，线程池由(RUNNING 或 SHUTDOWN ) 转变为 STOP 状态。
TIDYING：
- SHUTDOWN 状态下，任务数为0，其他所有任务已终止，线程池会变为 TIDYING 状态，会执行 terminated() 方法。线程池中的 terminated() 方法是空实现，可以重写该方法进行相应的处理。
- 线程池在 SHUTDOWN 状态，任务队列为空且执行中任务为空，线程池就会由 SHUTDOWN 转变为 TIDYING 状态。
- 线程池在 STOP 状态，线程池中执行中任务为空时，就会由 STOP 转变为 TIDYING 状态。
TERMINATED：线程池彻底终止。线程池在 TIDYING 状态执行完 terminated() 方法就会由 TIDYING 转变为 TERMINATED 状态。

Excutes

newFixedThreadPool
- 创建一个固定的长度的线程池，每当提交一个任务就创建一个线程，知道达到线程池的最大数量，这时线程规模将不再变化，当线程发生未预期的错误而结束时，线程池会创建一个新的线程继续运行队列里的任务。
newSingleThreadExecutor
- 这是一个单线程的 Executor ，它创建单个工作线程来执行任务，如果这个线程异常结束，会创建一个新的来代替它；它的特点是能确保依照任务在队列中的顺序来串行执行。
newCachedThreadPool
- 创建一个可缓存的线程池，如果线程池的规模超过了处理需求，将自动回收空闲线程，而当需求正驾驶，则可以自动添加新线程，线程池的规模不存在任何限制。
newScheduledThreadPool
- 创建一个固定长度的线程池，而且以延迟或定时的方式来执行任务，类似于 Timer
newSingleThreadScheduledExecutor
- 单线程可执行周期性任务的线程池
newWorkStealingPool
- 任务窃取线程池，不保证执行顺序，适合任务耗时差异较大。线程池中有多个线程队列，有的线程队列中有大量的比较耗时的任务堆积，而有的线程队列却是空的，就存在有的线程处于饥饿状态，当一个线程处于饥饿状态时，它就会去其它的线程队列中窃取任务。解决饥饿导致的效率问题。默认创建的并行 level 是 CPU 的核数。主线程结束，即使线程池有任务也会立即停止。

为什么不推荐Executors

Executors工具类创建的线程池队列或线程默认为Integer.MAX_VALUE，容易堆积请求阿里巴巴Java开发手册：

FixedThreadPool 和 SingleThreadPool：允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM
CachedThreadPool：允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM

推荐使用ThreadPoolExecutor类根据实际需要自定义创建

ThreadPoolExecutor

七大参数

ThreadPoolExecutor类主要有以下七个参数：

int corePoolSize: 核心线程池大小，线程池中常驻线程的最大数量
int maximumPoolSize: 最大核心线程池大小(包括核心线程和非核心线程)
long keepAliveTime: 线程空闲后的存活时间（仅适用于非核心线程）
TimeUnit unit: 超时单位
BlockingQueue<Runnable> workQueue: 阻塞队列
ThreadFactory threadFactory: 线程工厂：创建线程的，一般默认
RejectedExecutionHandler handle: 拒绝策略

四大策略

拒绝策略就是当队列满时，线程如何去处理新来的任务。

AbortPolicy（中止策略）-默认

功能：当触发拒绝策略时，直接抛出拒绝执行的异常
使用场景：ThreadPoolExecutor中默认的策略就是AbortPolicy，由于ExecutorService接口的系列ThreadPoolExecutor都没有显示的设置拒绝策略，所以默认的都是这个。

CallerRunsPolicy（调用者运行策略）

功能：只要线程池没有关闭，就由提交任务的当前线程处理。
使用场景：一般在不允许失败、对性能要求不高、并发量较小的场景下使用。

DiscardPolicy（丢弃策略）

功能：直接丢弃这个任务，不触发任何动作
使用场景：提交的任务无关紧要，一般用的少。

DiscardOldestPolicy（弃老策略）

功能：抛弃下一个将要被执行的任务，相当于排队的时候把第一个人打死，然后自己代替
使用场景：发布消息、修改消息类似场景。当老消息还未执行，此时新的消息又来了，这时未执行的消息的版本比现在提交的消息版本要低就可以被丢弃了。

工作队列

ArrayBlockingQueue:使用数组实现的有界阻塞队列，特性先进先出
LinkedBlockingQueue:使用链表实现的阻塞队列，特性先进先出，可以设置其容量，默认为Interger.MAX_VALUE
PriorityBlockingQueue:使用平衡二叉树堆，实现的具有优先级的无界阻塞队列
DelayQueue:无界阻塞延迟队列，队列中每个元素均有过期时间，当从队列获取元素时，只有过期元素才会出队列。队列头元素是最块要过期的元素。
SynchronousQueue:一个不存储元素的阻塞队列，每个插入操作，必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态

运行流程

判断线程池里的核心线程是否都在执行任务？
- 否：调用/创建一个新的核心线程来执行任务
- 是：工作队列是否已满？
  - 否：将新提交的任务存储在工作队列里
  - 是：线程池里的线程数是否达到最大线程值？
    - 否：调用/创建一个新的非核心线程来执行任务
    - 是：执行线程池饱和策略

一个面试题

一个线程池 core 7； max 20 ， queue： 50， 100并发进来怎么分配的？

答：先有7个能直接得到执行，接下来把50个进入队列排队等候，在多开13个继续执行。现在 70 个被安排上了。剩下 30 个默认执行饱和策略。

执行任务

execute提交没有返回值，不能判断是否执行成功。只能提交一个Runnable的对象
submit会返回一个Future对象，通过Future的get()方法来获取返回值，submit提交线程可以吃掉线程中产生的异常，达到线程复用。当get()执行结果时异常才会抛出。原因是通过submit提交的线程，当发生异常时，会将异常保存，待future.get()时才会抛出。

关闭线程池

shutdown()：不再继续接收新的任务，执行完成已有任务后关闭
shutdownNow()：直接关闭，若果有任务尝试停止

线程池出现异常会发生什么？

线程出现异常，线程会退出，并重新创建新的线程执行队列里任务，不能复用线程
当业务代码的异常捕获了，线程就可以复用
使用ThreadFactory的UncaughtExceptionHandler保证线程的所有异常都能捕获（包括业务的异常），兜底的.如果提交方式用execute，不能复用线程
setUncaughtExceptionHandler+submit ：可以吃掉异常并复用线程（是吃掉，不报错）
setUncaughtExceptionHandler+submit+future.get() ：可以获取到异常并复用线程

最佳实践

提交线程的业务异常用try catch处理，保证线程不会异常退出
业务之外的异常我们不可预见的，创建线程池设置ThreadFactory的UncaughtExceptionHandler可以对未捕获的异常做保底处理，通过submit提交任务，可以吃掉异常并复用线程；想要捕获异常这时用future.get()

注：关于异常处理的相关案例，已在源码中，这里不做展示

实战1：结合CompletableFuture使用线程池

CompletableFuture，结合了Future的优点，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。
CompletableFuture可以传入自定义线程池,否则使用自己默认的线程池，我们习惯做法是自定义线程池，控制整个项目的线程数量，不使用自定义的线程池，做到可控可调

步骤1：声明一个线程池bean

application.properties

//尽量做到每个业务使用自己配置的线程池
service1.thread.coreSize=10
service1.thread.maxSize=100
service1.thread.keepAliveTime=10
复制代码

线程池属性类

/**
 * @Description: 线程池属性
 * @Author: jianweil
 * @date: 2021/12/9 10:44
 */
@ConfigurationProperties(prefix = "service1.thread")
@Data
public class ThreadPoolConfigProperties {
    private Integer coreSize;
    private Integer maxSize;
    private Integer keepAliveTime;
}
复制代码

线程池配置类

/**
 * @Description: 线程池配置类:根据不同业务定义不同的线程池配置
 **/
@EnableConfigurationProperties(ThreadPoolConfigProperties.class)
@Configuration
public class MyService1ThreadConfig {

    @Bean
    public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {
        return new ThreadPoolExecutor(
                pool.getCoreSize(),
                pool.getMaxSize(),
                pool.getKeepAliveTime(),
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(100000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

}
复制代码

步骤2：使用

注：本文所有源码已分享github

/**
 * @Description: 测试CompletableFuture
 * @Author: jianweil
 * @date: 2021/12/9 10:50
 */
@SpringBootTest
public class CompletableFutureTest {

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    
    /***
     * 无返回值
     *  runAsync
     */
    @Test
    public void main1() {
        System.out.println("main.................start.....");
        CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, threadPoolExecutor);
        System.out.println("main.................end......");
    } 
}
复制代码

实战2：结合@Async使用线程池

在现实的互联网项目开发中，针对高并发的请求，一般的做法是高并发接口单独线程池隔离处理。可能为某一高并发的接口单独一个线程池

方式1：默认线程池

使用@Async注解，在默认情况下用的是SimpleAsyncTaskExecutor线程池，该线程池不是真正意义上的线程池
使用此线程池无法实现线程重用，每次调用都会新建一条线程。若系统中不断的创建线程，最终会导致系统占用内存过高，引发OutOfMemoryError错误

步骤1:自定义一个能查看线程池参数的类

不清楚线程池当时的情况，有多少线程在执行，多少在队列中等待呢？
创建了一个ThreadPoolTaskExecutor的子类，在每次提交线程任务的时候都会将当前线程池的运行状况打印出来

public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {

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

    private void showThreadPoolInfo(String prefix) {
        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();

        if (null == threadPoolExecutor) {
            return;
        }

        logger.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",
                this.getThreadNamePrefix(),
                prefix,
                threadPoolExecutor.getTaskCount(),
                threadPoolExecutor.getCompletedTaskCount(),
                threadPoolExecutor.getActiveCount(),
                threadPoolExecutor.getQueue().size());
    }

    @Override
    public void execute(Runnable task) {
        showThreadPoolInfo("1. do execute");
        super.execute(task);
    }

    @Override
    public void execute(Runnable task, long startTimeout) {
        showThreadPoolInfo("2. do execute");
        super.execute(task, startTimeout);
    }

    @Override
    public Future<?> submit(Runnable task) {
        showThreadPoolInfo("1. do submit");
        return super.submit(task);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        showThreadPoolInfo("2. do submit");
        return super.submit(task);
    }

    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
        showThreadPoolInfo("1. do submitListenable");
        return super.submitListenable(task);
    }

    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        showThreadPoolInfo("2. do submitListenable");
        return super.submitListenable(task);
    }
}
复制代码

步骤2:实现AsyncConfigurer类

要配置默认的线程池，要实现AsyncConfigurer类的两个方法
不需要打印运行状况的可以使用ThreadPoolTaskExecutor类构建线程池

/**
 * @Description: 注解@async配置
 * @Author: jianweil
 * @date: 2021/12/9 11:52
 */
@Slf4j
@EnableAsync
@Configuration
public class AsyncThreadConfig implements AsyncConfigurer {
    /**
     * 定义@Async默认的线程池
     * ThreadPoolTaskExecutor不是完全被IOC容器管理的bean,可以在方法上加上@Bean注解交给容器管理,这样可以将taskExecutor.initialize()方法调用去掉，容器会自动调用
     *
     * @return
     */
    @Override
    public Executor getAsyncExecutor() {
        int processors = Runtime.getRuntime().availableProcessors();
        //常用的执行器
        //ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //可以查看线程池参数的自定义执行器
        ThreadPoolTaskExecutor taskExecutor = new VisiableThreadPoolTaskExecutor();
        //核心线程数
        taskExecutor.setCorePoolSize(1);
        taskExecutor.setMaxPoolSize(2);
        //线程队列最大线程数,默认：50
        taskExecutor.setQueueCapacity(50);
        //线程名称前缀
        taskExecutor.setThreadNamePrefix("default-ljw-");
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化(重要)
        taskExecutor.initialize();
        return taskExecutor;
     }

    /**
     * 异步方法执行的过程中抛出的异常捕获
     *
     * @return
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) ->
                log.error("线程池执行任务发送未知错误,执行方法：{}", method.getName(), ex.getMessage());
    }

}
复制代码

步骤3: 使用

直接添加注解@Async即可使用到配置的默认线程池

/**
 * 默认线程池
 */
@Async
public void defaultThread() throws Exception {
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(1000));
    long end = System.currentTimeMillis();
    int i = 1 / 0;
    log.info("使用默认线程池，耗时：" + (end - start) + "毫秒");
}
复制代码

方式2：指定线程池

由于业务需要，根据业务不同需要不同的线程池

步骤1：声明一个线程池bean

/**
 * @Description: 注解@async配置
 * @Author: jianweil
 * @date: 2021/12/9 11:52
 */
@Slf4j
@EnableAsync
@Configuration
public class AsyncThreadConfig implements AsyncConfigurer {

    @Bean("service2Executor")
    public Executor service2Executor() {
        //Java虚拟机可用的处理器数
        int processors = Runtime.getRuntime().availableProcessors();
        //定义线程池
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //可以查看线程池参数的自定义执行器
        //ThreadPoolTaskExecutor taskExecutor = new VisiableThreadPoolTaskExecutor();
        //核心线程数
        taskExecutor.setCorePoolSize(processors);
        taskExecutor.setMaxPoolSize(100);
        //线程队列最大线程数,默认：100
        taskExecutor.setQueueCapacity(100);
        //线程名称前缀
        taskExecutor.setThreadNamePrefix("my-ljw-");
        //线程池中线程最大空闲时间，默认：60，单位：秒
        taskExecutor.setKeepAliveSeconds(60);
        //核心线程是否允许超时，默认:false
        taskExecutor.setAllowCoreThreadTimeOut(false);
        //IOC容器关闭时是否阻塞等待剩余的任务执行完成，默认:false（必须设置setAwaitTerminationSeconds）
        taskExecutor.setWaitForTasksToCompleteOnShutdown(false);
        //阻塞IOC容器关闭的时间，默认：10秒（必须设置setWaitForTasksToCompleteOnShutdown）
        taskExecutor.setAwaitTerminationSeconds(10);
        /**
         * 拒绝策略，默认是AbortPolicy
         * AbortPolicy：丢弃任务并抛出RejectedExecutionException异常
         * DiscardPolicy：丢弃任务但不抛出异常
         * DiscardOldestPolicy：丢弃最旧的处理程序，然后重试，如果执行器关闭，这时丢弃任务
         * CallerRunsPolicy：执行器执行任务失败，则在策略回调方法中执行任务，如果执行器关闭，这时丢弃任务
         */
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return taskExecutor;
    }
}
复制代码

步骤2: 使用

@Async("service2Executor")注解指定使用的线程池名称

/**
 * 指定线程池service2Executor
 *
 * @throws Exception
 */
@Async("service2Executor")
public void service2Executor() throws Exception {
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(1000));
    long end = System.currentTimeMillis();
    log.info("使用线程池service2Executor，耗时：" + (end - start) + "毫秒");
}
复制代码

注：异步任务返回值为void，不能获取的返回值的

计算线程数量

适合框架类

例如netty,dubbo这种底层通讯框架通常会参考进行设置

IO 密集型（较多）: 通常设置为2n+1，其中n为CPU核数
CPU 密集型（较少）: 通常设置为 n+1

实际情况往往复杂的多，并不会按照这个进行设置

IO密集型类型进阶算法

对于IO密集型类型的应用：线程数 = CPU核心数/(1-阻塞系数)
引入了阻塞系数的概念，一般为0.8~0.9之间，

在我们的业务开发中，基本上都是IO密集型，因为往往都会去操作数据库，访问redis，es等存储型组件，涉及到磁盘IO，网络IO。

一个4C8G的机器上部署了一个MQ消费者，在RocketMQ的实现中，消费端也是用一个线程池来消费线程的，那这个线程数要怎么设置呢？

如果按照 2n + 1 的公式，线程数设置为 9个，但在我们实践过程中发现如果增大线程数量，会显著提高消息的处理能力，说明 2n + 1 对于业务场景来说，并不太合适。
如果套用线程数 = CPU核心数/(1-阻塞系数) 阻塞系数取 0.8 ，线程数为20
如果我们发现数据库的操作耗时比较多，此时可以继续提高阻塞系数，从而增大线程数量。

那我们怎么判断需要增加更多线程呢？

其实可以用jstack命令查看一下进程的线程栈，如果发现线程池中大部分线程都处于等待获取任务，则说明线程够用
如果大部分线程都处于运行状态，可以继续适当调高线程数量。

线程数规划的公式（推荐）

《Java 并发编程实战》介绍了一个线程数计算的公式：

如果希望程序跑到CPU的目标利用率，需要的线程数公式为：

如果我期望目标利用率为90%（多核90），那么需要的线程数为：

把公式变个形，还可以通过线程数来计算CPU利用率：

虽然公式很好，但在真实的程序中，一般很难获得准确的等待时间和计算时间，因为程序很复杂，不只是“计算” 。一段代码中会有很多的内存读写，计算，I/O 等复合操作，精确的获取这两个指标很难，所以光靠公式计算线程数过于理想化。

真实程序中的线程数是没有固定答案，先设定预期，比如我期望的CPU利用率在多少，负载在多少，GC频率多少之类的指标后，再通过测试不断的调整到一个合理的线程数

获取CPU核心数

Java 获取CPU核心数

Runtime.getRuntime().availableProcessors()//获取逻辑核心数，如6核心12线程，那么返回的是12
复制代码

Linux 获取CPU核心数

# 总核数 = 物理CPU个数 X 每颗物理CPU的核数
# 总逻辑CPU数 = 物理CPU个数 X 每颗物理CPU的核数 X 超线程数

# 查看物理CPU个数
cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l

# 查看每个物理CPU中core的个数(即核数)
cat /proc/cpuinfo| grep "cpu cores"| uniq

# 查看逻辑CPU的个数
cat /proc/cpuinfo| grep "processor"| wc -l

本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。

THE END

java 后端开发语言数据结构程序人生程序员

二维码

)">

华为资深工程师总结的这本435页的书，居然把网络协议给趣谈了

< <上一篇

小公司怕倒闭，大厂怕裁员：现在的职场，程序员该何去何从？

下一篇>>

搜索内容

别告诉我你连线程池都不会用~ 一文搞懂线程池

线程池作用

线程池5种状态

Excutes

为什么不推荐Executors

ThreadPoolExecutor

七大参数

四大策略

AbortPolicy（中止策略）-默认

CallerRunsPolicy（调用者运行策略）

DiscardPolicy（丢弃策略）

DiscardOldestPolicy（弃老策略）

工作队列

运行流程

一个面试题

执行任务

关闭线程池

线程池出现异常会发生什么？

最佳实践

实战1：结合CompletableFuture使用线程池

步骤1：声明一个线程池bean

步骤2：使用

实战2：结合@Async使用线程池

方式1：默认线程池

步骤1:自定义一个能查看线程池参数的类

步骤2:实现AsyncConfigurer类

步骤3: 使用

方式2：指定线程池

步骤1：声明一个线程池bean

步骤2: 使用

计算线程数量

适合框架类

IO密集型类型进阶算法

线程数规划的公式（推荐）

获取CPU核心数

最新文章

分类

标签云