Java线程池ThreadPool详解

怼烎@ 2022-09-06 13:29 10阅读 0赞

### Java线程池ThreadPool详解 ###

*  1. 线程池简介
 *  2. 线程池特点
 *  3. 线程池核心参数
 *   *  3.1 核心参数
     *  3.2 拒绝策略
 *  4. 线程池工具类
 *  5. 线程池工作原理
 *  6. 不建议Executors创建线程池
 *  7. 手动创建线程池
 *  8. SpringBoot创建线程池

# 1. 线程池简介 #

线程池（英语： thread pool）：一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。

# 2. 线程池特点 #

线程池的优势： 线程池做的工作只要是控制运行的线程数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程数量超过了最大数量，超出数量的线程排队等候，等其他线程执行完毕，再从队列中取出任务来执行。

它的主要特点为：  
1.降低资源消耗: 通过重复利用已创建的线程降低线程创建和销毁造成的销耗。  
2.提高响应速度: 当任务到达时，任务可以不需要等待线程创建就能立即执行。  
3.提高线程的可管理性: 线程是稀缺资源，如果无限制的创建，不仅会销耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。  
4.Java 中的线程池是通过 Executor 框架实现的，该框架中用到了Executor， Executors，ExecutorService， ThreadPoolExecutor 这几个类。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70]

# 3. 线程池核心参数 #

## 3.1 核心参数 ##

• corePoolSize 线程池的核心线程数  
• maximumPoolSize 能容纳的最大线程数  
• keepAliveTime 空闲线程存活时间  
• unit 存活的时间单位  
• workQueue 存放提交但未执行任务的队列  
• threadFactory 创建线程的工厂类  
• handler 等待队列满后的拒绝策略

线程池中，有三个重要的参数，决定影响了拒绝策略：  
corePoolSize - 核心线程数，也即最小的线程数。  
workQueue - 阻塞队列 。  
maximumPoolSize -最大线程数。  
当提交任务数大于 corePoolSize 的时候，会优先将任务放到 workQueue 阻塞队列中。当阻塞队列饱和后，会扩充线程池中线程数，直到达到maximumPoolSize 最大线程数配置。此时，再多余的任务，则会触发线程池的拒绝策略了。总结起来，也就是一句话， 当提交的任务数大于（workQueue.size()+maximumPoolSize ），就会触发线程池的拒绝策略。

## 3.2 拒绝策略 ##

CallerRunsPolicy: 当触发拒绝策略，只要线程池没有关闭的话，则使用调用线程直接运行任务。一般并发比较小，性能要求不高，不允许失败。但是，由于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能效率上必然的损失较大。  
AbortPolicy: 丢弃任务，并抛出拒绝执行 RejectedExecutionException 异常信息。线程池默认的拒绝策略。必须处理好抛出的异常，否则会打断当前的执行流程，影响后续的任务执行。  
DiscardPolicy: 直接丢弃，其他啥都没有。  
DiscardOldestPolicy: 当触发拒绝策略，只要线程池没有关闭的话，丢弃阻塞队列 workQueue 中最老的一个任务，并将新任务加入。

# 4. 线程池工具类 #

线程池工具类，四种线程池创建方式  
Executors 是一个 Java 中的工具类。提供工厂方法来创建不同类型的线程池。Executors 的创建线程池的方法，创建出来的线程池都实现了ExecutorService 接口。常用方法有以下几个：

1.newFiexedThreadPool(int Threads)：创建固定数目线程的线程池。  
2.newCachedThreadPool()：创建一个可缓存的线程池，调用 execute将重用以前构造的线程（如果线程可用）。如果没有可用的线程，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。  
3.newSingleThreadExecutor() 创建一个单线程化的 Executor。  
4.newScheduledThreadPool(int corePoolSize) 创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代 Timer 类。

# 5. 线程池工作原理 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 1]  
1.在创建了线程池后，线程池中的线程数为零。  
2.当调用 execute()方法添加一个请求任务时，线程池会做出如下判断：  
2.1 如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；  
2.2 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列；  
2.3 如果这个时候队列满了且正在运行的线程数量还小于maximumPoolSize，那么还是要创建非核心线程立刻运行这个任务； 2.4 如果队列满了且正在运行的线程数量大于或等于maximumPoolSize，那么线程池会启动饱和拒绝策略来执行。

4.当一个线程完成任务时，它会从队列中取下一个任务来执行  
5.当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断：  
4.1 如果当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。  
4.2所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

# 6. 不建议Executors创建线程池 #

不建议大家使用Executors这个类来创建线程池呢，阿里开发手册这样定义：  
【强制】线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。说明： Executors 返回的线程池对象的弊端如下：  
1） FixedThreadPool 和 SingleThreadPool:  
允许的请求队列长度为 Integer.MAX\_VALUE，可能会堆积大量的请求，从而导致 OOM。  
2） CachedThreadPool 和 ScheduledThreadPool:  
允许的创建线程数量为 Integer.MAX\_VALUE， 可能会创建大量的线程，从而导致 OOM。

项目中创建多线程时，使用常见的三种线程池创建方式，单一、可变、定长都有一定问题，原因是 FixedThreadPool 和 SingleThreadExecutor 底层都是用LinkedBlockingQueue 实现的，这个队列最大长度为 Integer.MAX\_VALUE，容易导致 OOM。  
所以实际生产一般自己通过 ThreadPoolExecutor 的 7 个参数，自定义线程池。

# 7. 手动创建线程池 #

package com.zrj.unit.juc;
    
    import com.google.common.util.concurrent.ThreadFactoryBuilder;
    import org.junit.Test;
    
    import java.util.concurrent.*;
    
    /** * 线程池 * 注意：ThreadFactoryBuilder这里需要运用guava包，自定义线程名称 * * @author zrj * @since 2021/8/20 **/
    public class ThreadPoolExecutorTest { 
        //线程池的核心线程数
        private static int corePoolSize = 30;
        //能容纳的最大线程数
        private static int maximumPoolSize = 200;
        //空闲线程存活时间
        private static long keepAliveTime = 0L;
        //空闲线程存活时间 单位
        private static TimeUnit unit = TimeUnit.MILLISECONDS;
        //创建线程的工厂类,自定义线程名称
        private static String threadName = "thread-local-pool-%d";
        private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(threadName).build();
        //存放提交但未执行任务的队列
        private static BlockingQueue<Runnable> threadFactory = new LinkedBlockingQueue<>(1024);
        //等待队列满后的拒绝策略
        private static RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
        //定义线程池
        private static ExecutorService pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, threadFactory, namedThreadFactory, handler);
    
        // 定义售票窗口数
        private static int saleWindows = 10;
        // 自定义线程名称
        private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
        // 倒计时锁存器
        private static CountDownLatch latch = new CountDownLatch(saleWindows);
    
        /** * 模拟抢票 */
        @Test
        public void saleTicket() { 
            try { 
                for (int i = 0; i < saleWindows; i++) { 
                    pool.execute(() -> { 
                        try { 
                            String threadName = Thread.currentThread().getName();
                            System.out.println(threadName + "窗口开始售票");
                            Thread.sleep(1000);
                            System.out.println(threadName + "窗口售票完成");
                        } catch (InterruptedException e) { 
                            System.out.println("售票异常：" + e);
                        } finally { 
                            latch.countDown();
                        }
                    });
                }
            } catch (Exception e) { 
                System.out.println("系统异常：" + e);
            } finally { 
                pool.shutdown();
            }
    
            // 等待所有线程到达放行
            try { 
                latch.await();
            } catch (Exception e) { 
                System.out.println("系统异常，" + e);
            }
            System.out.println("所有线程执行完成，继续执行主线程");
        }
    }

# 8. SpringBoot创建线程池 #

springboot创建线程池，Spring提供的对ThreadPoolExecutor封装的线程池ThreadPoolTaskExecutor，直接使用注解启用。  
Async相当于是方法级别的线程，本身没有自定义线程池更加灵活  
相当于是每进来一个请求就开启一个线程，超过核心线程数小于最大线程数放入队列，队列满了，继续创建线程直至达到最大线程数。

application.properties

# 异步线程配置
    # 配置核心线程数
    async.executor.thread.core_pool_size = 5
    # 配置最大线程数
    async.executor.thread.max_pool_size = 5
    # 配置队列大小
    async.executor.thread.queue_capacity = 99999
    # 配置线程池中的线程的名称前缀
    async.executor.thread.name.prefix = async-service-

ExecutorConfig

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * 线程池的配置 * * @author zrj * @since 2021/9/9 **/
    @Slf4j
    @EnableAsync
    @Configuration
    public class ExecutorConfig { 
    
        @Value("${async.executor.thread.core_pool_size}")
        private int corePoolSize;
        @Value("${async.executor.thread.max_pool_size}")
        private int maxPoolSize;
        @Value("${async.executor.thread.queue_capacity}")
        private int queueCapacity;
        @Value("${async.executor.thread.name.prefix}")
        private String namePrefix;
    
        @Bean(name = "asyncServiceExecutor")
        public Executor asyncServiceExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //配置核心线程数
            executor.setCorePoolSize(corePoolSize);
            //配置最大线程数
            executor.setMaxPoolSize(maxPoolSize);
            //配置队列大小
            executor.setQueueCapacity(queueCapacity);
            //配置线程池中的线程的名称前缀
            executor.setThreadNamePrefix(namePrefix);
    
            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //执行初始化
            executor.initialize();
            return executor;
        }
    }

AsyncService

/** * @author zrj * @since 2021/9/9 **/
    public interface AsyncService { 
        void executeAsync();
    }

AsyncServiceImpl

package com.zrj.unit.service.impl;
    
    import com.zrj.unit.service.AsyncService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    /** * 线程服务 * Async相当于是方法级别的线程，本身没有自定义线程池更加灵活 * 相当于是每进来一个请求就开启一个线程，超过核心线程数小于最大线程数放入队列， * 队列满了，继续创建线程直至达到最大线程数 * * @author zrj * @since 2021/9/9 **/
    @Slf4j
    @Service
    public class AsyncServiceImpl implements AsyncService { 
    
        @Override
        @Async("asyncServiceExecutor")
        public void executeAsync() { 
            log.info( "start executeAsync" );
            try { 
                System.out.println( "异步线程要做的事情" );
                System.out.println( "可以在这里执行批量插入等耗时的事情" );
                Thread.sleep( 1000 );
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            log.info( "end executeAsync" );
        }
    }

AsyncController

package com.zrj.unit.controller;
    
    import com.zrj.unit.service.AsyncService;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    /** * 线程控制器 * * @author zrj * @since 2021/8/30 **/
    @RestController
    @RequestMapping("/async")
    public class AsyncController { 
    
        @Resource
        private AsyncService asyncService;
    
        /** * 多线程 */
        @GetMapping("/executor")
        public String executor() { 
            //开30个线程
            for (int i = 0; i < 30; i++) { 
                asyncService.executeAsync();
            }
            return "executor success";
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70]: /images/20220829/46ce6a77bb6242bfa42514fd44f73780.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTgzNjU1_size_16_color_FFFFFF_t_70 1]: /images/20220829/02c3929810204ee4844597a259e6182e.png