【线程池】四种创建方式

「爱情、让人受尽委屈。」 2024-04-19 11:04 75阅读 0赞

#### 线程池简介 ####

1.  线程池（英语：thread pool）：线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。 例如，线程数一般取cpu数量+2比较合适，线程数过多会导致额外的线程切换开销。
2.  Java中的两种线程池：  
    普通线程池：ThreadPoolExecutor  
    定时线程池：ScheduledThreadPoolExecutor

#### 1.1 ThreadPoolExecutor讲解： ####

1.  ThreadPoolExecutor是 java.util.concurrent 并发包下的类
2.  ThreadPoolExecutor 的构造方法共有4中：

public class ThreadPoolExecutor extends AbstractExecutorService {
    	
    	// 省略其他代码
    	
    	/**
         * 方法1：
         *
         * @param corePoolSize 核心线程数
         * @param maximumPoolSize 最大线程数
         * @param keepAliveTime 线程存活时间
         * @param unit  时间类型
         * @param workQueue 任务队列
         */
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            // 调用方法4
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                    Executors.defaultThreadFactory(), defaultHandler);
        }
    
        /**
         * 方法2：
         *
         * @param corePoolSize 核心线程数
         * @param maximumPoolSize 最大线程数
         * @param keepAliveTime 线程存活时间
         * @param unit 时间类型
         * @param workQueue 任务队列
         * @param threadFactory 创建新的线程Factory
         */
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory) {
            // 调用方法4
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                    threadFactory, defaultHandler);
        }
    
        /**
         * 方法3：
         * @param corePoolSize 核心线程数
         * @param maximumPoolSize 最大线程数
         * @param keepAliveTime 线程存活时间
         * @param unit 时间类型
         * @param workQueue 任务队列
         * @param handler 拒绝策略
         */
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  RejectedExecutionHandler handler) {
            // 调用方法4
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                    Executors.defaultThreadFactory(), handler);
        }
    
        /**
         * 方法4：
         * @param corePoolSize 核心线程数
         * @param maximumPoolSize 最大线程数
         * @param keepAliveTime 线程存活时间
         * @param unit 时间类型
         * @param workQueue 任务队列
         * @param threadFactory 创建新的线程Factory
         * @param handler 拒绝策略
         */
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) {
            if (corePoolSize < 0 ||
                    maximumPoolSize <= 0 ||
                    maximumPoolSize < corePoolSize ||
                    keepAliveTime < 0)
                throw new IllegalArgumentException();
            if (workQueue == null || threadFactory == null || handler == null)
                throw new NullPointerException();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }
    	
    	// 省略其他代码

ThreadPoolExecutor 中的其他3个构造方法都是最终调用了方法4，部分参数由ThreadPoolExecutor帮我们选择，那我们来详细看一下 方法4中参数介绍：

1、corePoolSize： 线程池中核心线程数，当新任务添加时，线程池中当前线程数（下面用 currentPoolSize 代替）

*  currentPoolSize < corePoolSize，通过threadFactory创建新的线程来执行任务。
 *  currentPoolSize >= corePoolSize 且 workQueue等待队列未满，则任务进入workQueue。
 *  currentPoolSize >= corePoolSize 且 workQueue等待队列已满 且 currentPoolSize < maximumPoolSize ，则创建新的线程执行任务
 *  currentPoolSize >= corePoolSize 且 workQueue等待队列已满 且 currentPoolSize >= maximumPoolSize，则调用拒绝策略 handler 来处理该任务。

2、maximumPoolSize：最大线程数，限制线程池中可以存在的最大线程数。

3、keepAliveTime：空闲线程的最大存活时间，如果 currentPoolSize > corePoolSize 且 workQueue 没有新的任务，当前线程在 keepAliveTime 时间内没有新的任务去执行，则会退出

4、unit ：对应 keepAliveTime 的时间类型：纳秒、微秒、毫秒、秒、分钟、小时、天。 java.util.concurrent.TimeUnit 中定义。

5、workQueue：保存需要执行的任务的队列。

6、threadFactory：线程池创建新的线程的工厂

7、handler ： 当任务超出队列 和 线程容量时执行策略

#### 1.2 ThreadPoolExecutor 常用的线程池： ####

1.  由于线程池 ThreadPoolExecutor 的创建需要N多参数，因为 JDK 为我们提供了 java.util.concurrent.Executors 其中帮我已经创建 3 中常用的 ThreadPoolExecutor 线程池，例：

public class Executors {
    
        // 省略部分代码
    
        /**
         * 1、创建一个可重用的，有固定数量的线程池
         * 2、每次提交一个任务就提交一个线程，直到达到线程达到线程池的大小，就不会创建线程了
         * 3、线程池的大小达到最大后稳定不变，如果一个线程异常终止，则会创建一个新的线程
         * @param nThreads
         * @return
         */
        public static ExecutorService newFixedThreadPool(int nThreads) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
        }
    
        /**
         * 创建只有一个线程的线程池
         * @return
         */
        public static ExecutorService newSingleThreadExecutor() {
            return new FinalizableDelegatedExecutorService
                    (new ThreadPoolExecutor(1, 1,
                            0L, TimeUnit.MILLISECONDS,
                            new LinkedBlockingQueue<Runnable>()));
        }
    
    
        /**
         * 创建数量不固定的线程池
         * 具有缓冲功能的线程池，系统根据需要创建线程，线程会被缓冲到线程中
         * 如果线程池大小超过了处理任务需要的线程线程池就会回收空闲的线程池，
         * 当处理任务增加时，线程池可以增加线程来处理任务线程池不会对线程的大
         * 小进行限制线程池
         */
        public static ExecutorService newCachedThreadPool() {
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                    60L, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>());
        }
    
        // 装饰线程池
        static class FinalizableDelegatedExecutorService extends Executors.DelegatedExecutorService {
            FinalizableDelegatedExecutorService(ExecutorService executor) {
                super(executor);
            }
    
            /**
             * 1、当对象不再被任何对象引用时，GC会调用该对象的finalize()方法
             * 2、保证线程池被回收时任务能正常执行结束
             */
            protected void finalize() {
                super.shutdown();
            }
        }
    
        // 省略部分代码
    
    }

根据Executors 中代码看出，类中定义了不同的方法，通过创建不同的ThreadPoolExecutor来实现线程池，我们结合 1.1 中的讲解以及注释已不难理解，不过多解读了。下面我们在实际测试上面的三种方法：

/**
     * @describe: 常用默认线程池 学习
     *
     */
    public class CommonPool {
    
        public static void main(String[] args) {
           CommonPool commonPool = new CommonPool();
           commonPool.singlePoolTest();
           commonPool.newThreadpool();
            commonPool.CachedPool();
        }
    
        /**
         * 创建只有一个线程的线程池
         */
        public  void singlePoolTest(){
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            for (int i =0; i<1000; i++){
                executorService.execute(new RunableTest() );
            }
            executorService.shutdown();
        }
    
    
        /**
         * 1、创建一个可重用的，有固定数量的线程池
         * 2、每次提交一个任务就提交一个线程，直到达到线程达到线程池的大小，就不会创建线程了
         * 3、线程池的大小达到最大后稳定不变，如果一个线程异常终止，则会创建一个新的线程
         */
        public void newThreadpool(){
            Integer poolNum = 2;
            ExecutorService executorService = Executors.newFixedThreadPool(poolNum);
            for (int i =0; i<1000; i++){
                executorService.execute(new RunableTest() );
            }
        }
    
        /**
         * 创建数量不固定的线程池
         * 具有缓冲功能的线程池，系统根据需要创建线程，线程会被缓冲到线程中
         * 如果线程池大小超过了处理任务需要的线程线程池就会回收空闲的线程池，
         * 当处理任务增加时，线程池可以增加线程来处理任务线程池不会对线程的大
         * 小进行限制线程池
         */
        public void CachedPool(){
            ExecutorService executorService = Executors.newCachedThreadPool();
            for (int i =0; i<1000; i++){
                executorService.execute(new RunableTest() );
            }
        }
        class RunableTest implements Runnable{
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName());
            }
        }
    }

#### 2.1 ScheduledThreadPoolExecutor讲解： ####

1.  在创建线程池时，默认也是调用 ThreadPoolExecutor 来实现的，只不过在任务队列中其使用了 DelayedWorkQueue 延时队列来实现任务的延时功能【有兴趣的小伙伴可以在研究一下】：  
    定时线程池例：

public class Test {
    
        public static void main(String[] args) {
            // 创建一个 核心线程数为 5 的定时线程池
            ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
            for (int i = 0; i < 10; i++) {
                // 任务 3 秒后会执行
                scheduledThreadPool.schedule(new Runnable() {
                    public void run() {
                        System.out.println("delay 3 seconds");
                    }
                }, 3, TimeUnit.SECONDS);
            }
        }
    }

创建定时线程池源码：

// Executors 中代码：
       public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
            return new ScheduledThreadPoolExecutor(corePoolSize);
        }
    
     // ScheduledThreadPoolExecutor 中代码：
      public ScheduledThreadPoolExecutor(int corePoolSize) {
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }
      // ThreadPoolExecutor 中代码：
      public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }

#### 常用线程池队列简介： ####

1.  **ArrayBlockingQueue** ：  
    a、ArrayBlockingQueue是一个有界缓存等待队列，可以指定缓存队列的大小，当正在执行的线程数等于corePoolSize时，多余的元素缓存在ArrayBlockingQueue队列中等待有空闲的线程时继续执行，当ArrayBlockingQueue已满时，加入ArrayBlockingQueue失败，会开启新的线程去执行，当线程数已经达到最大的maximumPoolSizes时，再有新的元素尝试加入ArrayBlockingQueue时会报错。  
    b、ArrayBlockingQueue 推荐博文：[https://blog.csdn.net/zhangyong01245/article/details/102390093][https_blog.csdn.net_zhangyong01245_article_details_102390093]
2.  **LinkedBlockingQueue**：  
    a、LinkedBlockingQueue是一个无界缓存等待队列。当前执行的线程数量达到corePoolSize的数量时，剩余的元素会在阻塞队列里等待。（所以在使用此阻塞队列时maximumPoolSizes就相当于无效了），每个线程完全独立于其他线程。生产者和消费者使用独立的锁来控制数据的同步，即在高并发的情况下可以并行操作队列中的数据。  
    b、LinkedBlockingQueue推荐博文： [https://blog.csdn.net/zhangyong01245/article/details/102396551][https_blog.csdn.net_zhangyong01245_article_details_102396551]
3.  **SynchronousQueue：**  
    a、SynchronousQueue没有容量，是无缓冲等待队列，是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。  
    b、拥有公平（FIFO）和非公平(LIFO)策略，非公平策略会导致一些数据永远无法被消费的情况？  
    c、使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作。  
    d、SynchronousQueue推荐博文： [https://blog.csdn.net/zhangyong01245/article/details/102483284][https_blog.csdn.net_zhangyong01245_article_details_102483284]
4.  DelayedWorkQueue 推荐博文： [https://blog.csdn.net/nobody\_1/article/details/99684009][https_blog.csdn.net_nobody_1_article_details_99684009]

## 另外： ##

线程池中拒绝策略：[https://blog.csdn.net/zhangyong01245/article/details/101002180][https_blog.csdn.net_zhangyong01245_article_details_101002180]

#### 重点： ####

有使用过 Alibaba 开发规范的小伙伴们肯定会注意到 上述的创建线程池的方法 是不建议的。Alibab理由如下：

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

#### 工作中常用： ####

1.  实际工作中定义核心线程数：  
    a. 当我们的任务属于 IO密集型，操作文件或者http时： 线程数 =CPU \* 2  
    b.但我们的任务属于计算密集型时： 线程数 = CPU +1
2.  在实际项目开发中 我们既不会用 Executors 也不会用 ThreadPoolExecutor 来直接创建线程池，而是使用 Spring 提供的包装类 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor 来创建线程池，  
    XML 示例如下：

<bean id="executor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    		
    		<property name="corePoolSize" value="2" />
    		
    		<property name="keepAliveSeconds" value="200" />
    		
    		<property name="maxPoolSize" value="15" />
    		
    		<property name="queueCapacity" value="20" />
    		
    		<property name="rejectedExecutionHandler">
    			<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
    		</prope

Java 部分代码如下:

public class XxxxxService {
    
        // 利用Spring的自动注入功能注入
        @Resource(name = "executor")
        private ThreadPoolTaskExecutor executor;
    
        public void xxxx(){
            // 业务代码 省略
    
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println("Hello World");;
                }
            });
        }
    }

## END !!! ##

[https_blog.csdn.net_zhangyong01245_article_details_102390093]: https://blog.csdn.net/zhangyong01245/article/details/102390093
[https_blog.csdn.net_zhangyong01245_article_details_102396551]: https://blog.csdn.net/zhangyong01245/article/details/102396551
[https_blog.csdn.net_zhangyong01245_article_details_102483284]: https://blog.csdn.net/zhangyong01245/article/details/102483284
[https_blog.csdn.net_nobody_1_article_details_99684009]: https://blog.csdn.net/nobody_1/article/details/99684009
[https_blog.csdn.net_zhangyong01245_article_details_101002180]: https://blog.csdn.net/zhangyong01245/article/details/101002180