线程池系列(一)------------线程池简介

蔚落 2021-12-10 23:59 438阅读 0赞

# 一、简介 #

线程池主要解决了俩个问题：一是避免了频繁创建销毁线程所带来的开销，达到了线程复用的目的；二是线程池提供了一种资源限制和管理的手段，用户可以通过一些参数来定制化线程；

# 二、线程池状态 #

## 2.1 原子变量ctl ##

### 2.1.1 ThreadPoolExecutor源码 ###

public class ThreadPoolExecutor extends AbstractExecutorService {
    
        // 该原子变量的高3位用来表示线程池状态，低29位用来表示线程个数(假设Integer为32位时)， 默认是RUNNING状态
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    
        /**
         * 线程个数掩码位数，并非所有平台的int类型都是32位的，所以准确的来说，是具体平台下Integer的二进制位数-3后的剩余位数所表示的数才是线程的个数
         */
        private static final int COUNT_BITS = Integer.SIZE - 3;
    
        
        // 线程的最大个数（低29位 00011111111111111111111111111111）
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
        
        // RUNNING状态 高三位为111  （11100000000000000000000000000000）
        private static final int RUNNING    = -1 << COUNT_BITS;
       
        // SHUTDOWN状态 高三位为000 （00000000000000000000000000000000）
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
    
        // STOP状态  高三位为001    （00100000000000000000000000000000）
        private static final int STOP       =  1 << COUNT_BITS;
    
        // TIDYING状态 高三位为010  （01000000000000000000000000000000）
        private static final int TIDYING    =  2 << COUNT_BITS;
    
        // TERMINATED状态 高三位为011（01100000000000000000000000000000）
        private static final int TERMINATED =  3 << COUNT_BITS;
        
        
        // 获取高3位(运行状态)
        private static int runStateOf(int c)     { return c & ~CAPACITY; }
       
        // 获取低29位(线程个数)
        private static int workerCountOf(int c)  { return c & CAPACITY; }
    
        // 计算ctl新值（线程状态和线程个数）
        private static int ctlOf(int rs, int wc) { return rs | wc; }
    
    
    }

## 2.2 状态转换图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MjA2MjYx_size_16_color_FFFFFF_t_70][]

## 2.3 状态详解 ##

### 2.3.1 RUNNING ###

*  接受新任务并且处理阻塞队列里的任务；

### 2.3.2 SHUTDOWN ###

*  拒绝新任务但是处理阻塞队列里的任务；

### 2.3.3 STOP ###

*  拒绝新任务并且抛弃阻塞队列里的任务，同时会中断正在处理的任务；

### 2.3.4 TIDYING ###

*  所有任务都执行完（包含阻塞队列里面的任务）后当前线程池活动线程数为0-，将调用terminated方法；

### 2.3.5 TERMINATED ###

*  终止状态。terminated方法调用完成以后的状态。

# 三、线程池七大核心参数 #

## 3.1 corePoolSize ##

*  核心线程个数；

## 3.2 wordQueue ##

*  用于保存等待执行的任务的阻塞队列，有以下几种：
    
     *  ArrayBlockingQueue: 基于数组的有界阻塞队列；
     *  LinedBlockingQueue: 基于链表的无界阻塞队列；
     *  SynchronousQueue： 最多只有一个元素的同步队列；
     *  PriorityBlockingQueue: 优先级队列；
     *  DelayQueue: 无界阻塞延迟队列；

## 3.3 maximunPoolSize ##

*  线程池最大线程数量；

## 3.4 ThreadFactory ##

*  创建线程的工厂；

## 3.5 RejectidExecutionHandler ##

*  饱和策略/拒绝策略，当队列满并且县城个数达到maximunPoolSize后采取的策略，有以下几种：
    
     *  AbortPolicy： 抛出异常；
     *  DiscardPolicy: 默默丢弃，不抛出异常；
     *  DiscardOldestPolicy： 调用poll丢弃一个任务，执行当前任务；
     *  CallerRunsPolicy: 使用调用者所在线程来运行任务；

## 3.6 keeyAliveTime ##

*  存活时间。如果当前线程池中的线程数量比核心线程数量多，并且是闲置状态，则这些闲置的线程能存活的最大时间；

## 3.7 TimeUnit ##

*  存活时间的时间单位；

# 四、六大线程池类型 TODO #

## 4.1 FixedThreadPool ##

### 4.1.1 Executors类中源码 ###

*      /**
             *
             */   
            public static ExecutorService newFixedThreadPool(int nThreads) {
                return new ThreadPoolExecutor(nThreads, nThreads,
                                              0L, TimeUnit.MILLISECONDS,
                                              new LinkedBlockingQueue<Runnable>());
            }
        
            
            /**
             *
             */
            public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
                return new ThreadPoolExecutor(nThreads, nThreads,
                                              0L, TimeUnit.MILLISECONDS,
                                              new LinkedBlockingQueue<Runnable>(),
                                              threadFactory);
            }

### 4.1.2 详解 ###

## 4.2 CachedThreadPool ##

### 4.2.1 Executors类中源码 ###

*      /**
             *
             */
            public static ExecutorService newCachedThreadPool() {
                return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                              60L, TimeUnit.SECONDS,
                                              new SynchronousQueue<Runnable>());
            }
        
        
            /**
             *
             */
            public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
                return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                              60L, TimeUnit.SECONDS,
                                              new SynchronousQueue<Runnable>(),
                                              threadFactory);
            }

### 4.2.2 详解 ###

## 4.3 SingleThreadExecutor ##

### 4.3.1 Executors类中源码 ###

### 4.3.2 详解 ###

## 4.4 **ScheduledThreadpool** ##

### 4.4.1 Executors中源码 ###

### 4.4.2 详解 ###

## 4.5 **WorkStealingPool** ##

### 4.5.1 Executors类中源码 ###

*      /**
             *
             */
            public static ExecutorService newWorkStealingPool() {
                return new ForkJoinPool
                    (Runtime.getRuntime().availableProcessors(),
                     ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                     null, true);
            }
        
            
            /**
             *
             */
            public static ExecutorService newWorkStealingPool(int parallelism) {
                return new ForkJoinPool
                    (parallelism,
                     ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                     null, true);
            }

### 4.5.2 详解 ###

## 4.6 **ForkJoinPool** ##

### 4.6.1 **ForkJoinPool类**中源码 ###

*      /**
             *
             */
            public ForkJoinPool() {
                this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
                     defaultForkJoinWorkerThreadFactory, null, false);
            }
        
            
            /**
             *
             */
            public ForkJoinPool(int parallelism) {
                this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
            }
        
        
            /**
             *
             */
            public ForkJoinPool(int parallelism,
                                ForkJoinWorkerThreadFactory factory,
                                UncaughtExceptionHandler handler,
                                boolean asyncMode) {
                this(checkParallelism(parallelism),
                     checkFactory(factory),
                     handler,
                     asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
                     "ForkJoinPool-" + nextPoolId() + "-worker-");
                checkPermission();
            }
        
            
            /**
             *
             */
            private ForkJoinPool(int parallelism,
                                 ForkJoinWorkerThreadFactory factory,
                                 UncaughtExceptionHandler handler,
                                 int mode,
                                 String workerNamePrefix) {
                this.workerNamePrefix = workerNamePrefix;
                this.factory = factory;
                this.ueh = handler;
                this.config = (parallelism & SMASK) | mode;
                long np = (long)(-parallelism); // offset ctl counts
                this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
            }

### 4.6.2 详解 ###

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MjA2MjYx_size_16_color_FFFFFF_t_70]: /images/20211210/a034573bbaa54a89b015468b3102044b.png