带你走近线程池（一）：线程池工作机制和原理

小灰灰 2022-05-14 02:40 260阅读 0赞

# 一、为什么需要线程池 #

随着系统用户的逐渐增多，为了提高用户响应，提供一个高并发，高可用的系统。java的线程池就可以解决很多问题。很多异步，并发的场景都可以用到线程池。

1.降低重复创建线程的开销，统一管理线程。

2.提高响应速度，任务提交后，不需要等待线程创建的过程。

3.提高线程的可管理性。重复管理线程，避免创建大量的线程增加开销。

--------------------

# 二、线程池的处理流程 #

## 1.线程池构造参数 ##

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
       private final BlockingQueue<Runnable> workQueue;
       private final ReentrantLock mainLock = new ReentrantLock();
       private final HashSet<Worker> workers = new HashSet<Worker>();
       private volatile long keepAliveTime;
       private volatile boolean allowCoreThreadTimeOut;
       private volatile int corePoolSize;
       private volatile int maximumPoolSize;
       private volatile RejectedExcutionHandler 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;
        }

<table>  基本参数  
 <tbody> 
  <tr> 
   <th style="width:174px;"> <p>corePoolSize</p> </th> 
   <td style="width:675px;">核心运行的poolSize，也就是当超过这个范围的时候，就需要将新的Thread放入到等待队列中了</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">maximumPoolSize&nbsp;</th> 
   <td style="width:675px;">&nbsp;线程池最大容量大小，当大于了这个值就会将Thread由一个丢弃处理机制来处理</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">keepAliveTime&nbsp;</th> 
   <td style="width:675px;">默认都是0，当线程没有任务处理后，保持多长时间</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">TimeUnit&nbsp;</th> 
   <td style="width:675px;">时间单位，</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">ThreadFactory&nbsp;</th> 
   <td style="width:675px;">是构造Thread的方法，你可以自己去包装和传递，主要实现newThread方法即可</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">BlockingQueue</th> 
   <td style="width:675px;">等待队列，当达到corePoolSize的时候，就向该等待队列放入线程信息（默认为一个<strong>LinkedBlockingQueue</strong>），运行中的队列属性为：workers，为一个HashSet；内部被包装了一层</td> 
  </tr> 
  <tr> 
   <th style="width:174px;">RejectedExecutionHandler&nbsp;</th> 
   <td style="width:675px;">参数<strong>maximumPoolSize</strong>达到后丢弃处理的方法，java提供了4种丢弃处理的方法，当然你也可以自己弄，主要是要实现接口：RejectedExecutionHandler中的方法</td> 
  </tr> 
 </tbody> 
</table>

## 2.处理流程 ##

# ![70][] #

主要步骤：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程。然后将任务提交到队列，如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数就没什么效果。当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。

--------------------

# 三、ThreadPoolExecutor #

## 1.具体执行方法 ##

![70 1][]

有一个重要的变量：

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl维护两个概念上的参数：workCount和runState。

workCount表示有效的线程数量，

runState表示线程池的运行状态。

public void execute(Runnable command) {
            //检查command不能为null
            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();
            //如果当前线程小于corePoolSize
            if (workerCountOf(c) < corePoolSize) {
                //如果添加Worker线程成功，则返回
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            //如果当前正在运行阶段并且可以将线程入队
            if (isRunning(c) && workQueue.offer(command)) {
                //再次检查ctl状态
                int recheck = ctl.get();
                //如果不在运行状态了，那么就从队列中移除任务
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                //如果在运行阶段，但是Worker数量为0，调用addWorker方法
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            //如果不能入队，且不能创建Worker，那么reject
            else if (!addWorker(command, false))
                reject(command);
        }

以上是核心的执行任务的方法：

从注释中可以看到execute方法分为三步：  
（1）如果小于corePoolSize的线程正在运行，那么创建一个核心线程，并将任务作为核心线程的第一个任务  
（2）如果一个任务可以加入到队列中，然后需要在此检查是否需要新建一个线程（因为可能存在一个线程在上次检查完之后被回收了）或者因为线程池停止了。所以需要再次检查状态，如果不在RUNNING状态并且能够成功移除任务的话，那么调用reject方法，否则就调用addWorker(null,false)方法。  
（3）如果任务不能放入队列，会首先尝试添加一个新线程（非核心线程）。如果失败，则调用reject方法。

[70]: /images/20220514/bf023791cb594359a103be89f37e131a.png
[70 1]: /images/20220514/1c189ab1b71742db8db75821e122992e.png