Java 线程池源码详解(1)-基本线程池解析

蔚落 2024-04-18 16:56 95阅读 0赞

#### 文章目录 ####

*  1. 线程池的使用
 *   *   *  1.1 创建线程池
         *  1.2 线程池核心 ThreadPoolExecutor
         *   *  1.2.1 ThreadPoolExecutor 构造方法
             *  1.2.1 ThreadPoolExecutor 关键方法
             *   *  1.2.1.1 线程池任务处理方法
                 *  1.2.1.2 线程池关闭方法
 *  2. 线程池任务处理原理
 *   *   *  2.1 线程池状态
         *  2.2 线程池任务执行流程
 *  3. 配置线程池
 *   *   *  3.1 配置线程数量
         *  3.2 线程池状态监控

## 1. 线程池的使用 ##

如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为创建线程和销毁线程都需要时间。

*  为了解决这种问题，一个思想是将线程资源放到一个池子中，每次使用都从里面获取，用完之后又放回池子供其他任务使用，从而达到线程复用的目的，也就是线程池技术。

#### 1.1 创建线程池 ####

JDK 1.5 之后推出创建线程池的相关 api，常见的创建线程池方式有以下：

*  `Executors.newCachedThreadPool()`：创建无限线程数量的线程池
 *  `Executors.newFixedThreadPool(nThreads)`：创建固定线程数量的线程池  
    以上两种线程池实际上都是通过 ThreadPoolExecutor 来创建的：
    
        public static ExecutorService newCachedThreadPool() {
              
              return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                            60L, TimeUnit.SECONDS,
                                            new SynchronousQueue<Runnable>());
          }
 *  `Executors.newScheduledThreadPool()`：创建可执行定时任务的线程池  
    定时任务线程池是通过 ScheduledThreadPoolExecutor来创建的：
    
        public static ScheduledExecutorService newScheduledThreadPool(
                  int corePoolSize, ThreadFactory threadFactory) {
              
              return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
          }
 *  `Executors.newSingleThreadExecutor()`：创建单个线程的线程池  
    单线程线程池通过 FinalizableDelegatedExecutorService来创建：
    
        public static ExecutorService newSingleThreadExecutor() {
              
              return new FinalizableDelegatedExecutorService
                  (new ThreadPoolExecutor(1, 1,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>()));
          }

以上四种线程池的继承结构及组合如下图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70]

#### 1.2 线程池核心 ThreadPoolExecutor ####

##### 1.2.1 ThreadPoolExecutor 构造方法 #####

在ThreadPoolExecutor类中提供了四个构造方法：

public class ThreadPoolExecutor extends AbstractExecutorService {
        
        .....
        public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
        TimeUnit unit,BlockingQueue<Runnable> workQueue);
     
        public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
        TimeUnit unit, BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
     
        public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
        TimeUnit unit, BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
     
        public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
        TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,
        RejectedExecutionHandler handler);
    }

构造方法中各个参数的含义：

1.  **corePoolSize**：线程池的基本大小  
    当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，直到线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中。 如果调用了线程池的 prestartAllCoreThreads() 方法或者 prestartCoreThread() 方法，线程池会提前创建并启动所有基本线程，此举可用来预热线程池
2.  **maximumPoolSize**：线程池最大线程大小  
    这是一个非常重要的参数，它表示在线程池中最多能创建多少个线程，如果任务队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。
3.  **keepAliveTime**：表示线程空闲后的存活时间  
    默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了 allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0
4.  **unit**：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：
    
        TimeUnit.DAYS;               //天
        TimeUnit.HOURS;             //小时
        TimeUnit.MINUTES;           //分钟
        TimeUnit.SECONDS;           //秒
        TimeUnit.MILLISECONDS;      //毫秒
        TimeUnit.MICROSECONDS;      //微妙
        TimeUnit.NANOSECONDS;       //纳秒
5.  **workQueue**：一个阻塞队列，用来存储等待执行的任务  
    这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择。  
    ArrayBlockingQueue 和 PriorityBlockingQueue 使用较少，一般使用 LinkedBlockingQueue 和Synchronous。线程池的排队策略与BlockingQueue有关。
    
        ArrayBlockingQueue：基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序
        
        LinkedBlockingQueue：基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue
        
        SynchronousQueue：不存储元素的阻塞队列
        
        PriorityBlockingQueue：具有优先级的无限阻塞队列
6.  **threadFactory**：线程工厂，主要用来创建线程
7.  **handler**：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是`AbortPolicy`，表示无法处理新任务时抛出异常。有以下四种取值：
    
        ThreadPoolExecutor.AbortPolicy: 丢弃任务并抛出RejectedExecutionException异常
        ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常
        ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
        ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

##### 1.2.1 ThreadPoolExecutor 关键方法 #####

###### 1.2.1.1 线程池任务处理方法 ######

1.  **ThreadPoolExecutor\#execute()**  
    实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是 ThreadPoolExecutor 的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。
2.  **ThreadPoolExecutor\#submit()**  
    是在 ExecutorService 中声明的方法，在 AbstractExecutorService 就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果 future 对象，通过这个future可判断任务是否执行成功。submit()方法的实现实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。
    
        public Future<?> submit(Runnable task) {
              
             if (task == null) throw new NullPointerException();
             RunnableFuture<Void> ftask = newTaskFor(task, null);
             execute(ftask); // 调用 execute 方法
             return ftask;
         }

###### 1.2.1.2 线程池关闭方法 ######

原理是遍历线程池中的工作线程，然后逐个调用线程的 interrupt 方法来中断线程，所以无法响应中断的任务可能永远无法终止。

1.  **ThreadPoolExecutor\#shutdown()**  
    不会立即终止线程池，而是将线程池的状态设置成**SHUTDOWN**，让正在执行的任务继续执行，但再也不会接受新的任务
    
    > 需注意此处让正在执行的任务继续执行并不意味着会等待任务执行完毕，如果要等待需要调用 `ThreadPoolExecutor#awaitTermination()` 方法
    
        public void shutdown() {
              
             final ReentrantLock mainLock = this.mainLock;
             mainLock.lock();
             try {
              
                 checkShutdownAccess();
                 advanceRunState(SHUTDOWN); // 设置线程池状态
                 interruptIdleWorkers(); // 中断未执行（IDLE）的任务
                 onShutdown(); // hook for ScheduledThreadPoolExecutor
             } finally {
              
                 mainLock.unlock();
             }
             tryTerminate();
         }
2.  **ThreadPoolExecutor\#shutdownNow()**  
    立即终止线程池，将线程池的状态设置成**STOP**并尝试中断正在执行的任务，清空任务缓存队列，返回尚未执行的任务
    
        public List<Runnable> shutdownNow() {
              
             List<Runnable> tasks;
             final ReentrantLock mainLock = this.mainLock;
             mainLock.lock();
             try {
              
                 checkShutdownAccess();
                 advanceRunState(STOP); // 设置线程池状态
                 interruptWorkers();  //  中断所有的任务
                 tasks = drainQueue();
             } finally {
              
                 mainLock.unlock();
             }
             tryTerminate();
             return tasks; // 返回未执行任务 list
         }

## 2. 线程池任务处理原理 ##

#### 2.1 线程池状态 ####

线程池中线程状态主要定义为以下：

// runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

*  `RUNNING` 运行状态，指可以接受任务执行队列里的任务
 *  `SHUTDOWN` 调用了 shutdown() 方法，不再接受新任务了，但是队列里的任务得执行完毕
 *  `STOP` 调用了 shutdownNow() 方法，不再接受新任务，同时抛弃阻塞队列里的所有任务并中断所有正在执行任务
 *  `TIDYING` 所有任务都执行完毕，在调用 shutdown()/shutdownNow() 中都会尝试更新为这个状态。
 *  `TERMINATED` 终止状态，当执行 terminated() 后会更新为这个状态。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70 1]

#### 2.2 线程池任务执行流程 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70_pic_center]

1.  在`ThreadPoolExecutor`类中，最核心的任务提交方法是`ThreadPoolExecutor#execute()`方法，该方法内调用`ThreadPoolExecutor#addWorker()`方法会新建线程并启动
2.  `Worker` 实现 `Runnable` 接口，保存了线程和 Runnable 任务作为成员变量，在其 `Worker#run()` 方法中调用 `ThreadPoolExecutor#runWorker()` 方法会调用 `ThreadPoolExecutor#getTask()` 方法从线程池的任务队列中获取任务，接着调用了 Runnable 任务的 `run()`方法完成任务处理

> 1.  获取当前线程池的状态
> 2.  当前线程数量小于 coreSize 时创建一个新的线程运行
> 3.  如果当前线程池处于运行状态，并且任务写入阻塞队列成功  
>     1. 双重检查，再次获取线程池状态；如果线程池状态变为非运行状态，从阻塞队列移除任务并尝试终止线程池，之后执行拒绝新任务的策略  
>     2. 如果当前线程池为空就新创建一个线程并执行
> 4.  如果在第 3 步的判断线程池为非运行状态，尝试新建线程，如果失败则执行拒绝策略  
>     ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70 2]  
>     需注意：
>     
>     1.  当线程池中线程执行任务的时候，任务出现未被捕获的异常的情况下，线程池会将运行该任务的线程从池中移除并销毁，且同时会创建一个新的线程加入到线程池中，这种情况由 `ThreadPoolExecutor#processWorkerExit()` 方法处理
>     2.  可以通过 ThreadFactory 自定义线程并捕获线程内抛出的异常，但是无论是否去捕获和处理线程池中工作线程抛出的异常，这个线程都会从线程池中被移除

public void execute(Runnable command) {
        
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) {
        
                if (addWorker(command, true)) // 处理任务
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {
        
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            else if (!addWorker(command, false))
                reject(command);
        }

## 3. 配置线程池 ##

#### 3.1 配置线程数量 ####

线程池中线程数量的配置需要根据任务执行的性质来确定。

*  IO 密集型任务：由于线程并不是一直在运行，所以可以尽可能的多配置线程，比如 CPU 个数 \* 2
 *  CPU 密集型任务（大量复杂的运算）应当分配较少的线程，比如 CPU 个数相当的线程数量

这些都是经验值，最好的方式还是根据实际情况测试得出最佳配置。

#### 3.2 线程池状态监控 ####

可以继承线程池扩展其中的几个函数来自定义监控逻辑，可以在线程执行前、后、终止状态执行自定义逻辑。

protected void beforeExecute(Thread t, Runnable r) {
         }
     protected void afterExecute(Runnable r, Throwable t) {
         }
     protected void terminated() {
         }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/f8fce42ba950405c9909fd77ac27eeb8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/eaafa3adad964ec48ba6bcdcbdd9b35d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/8aa7e2b44a524571951a3897e7930d1a.jpg
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTUwNTMxMw_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/15a0121e095148df966c134977227112.png