JAVA线程池源码详解（JDK1.8）

女爷i 2023-03-13 03:54 11阅读 0赞

# 线程池优点 #

1.  重用线程池中的线程，避免线程的创建销毁带来的性能开销。
2.  控制线程池的最大并发数。
3.  对线程简单管理：如定时执行以及指定间隔循环执行。

## 线程池五种状态 ##

在计算中，数据都是以补码存放的。

// 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;

**未左移之前的状态值：**

<table> 
 <thead> 
  <tr> 
   <th>状态值</th> 
   <th>二进制</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>RUNNING</td> 
   <td>11111111111111111111111111111111</td> 
  </tr> 
  <tr> 
   <td>SHUTDOWN</td> 
   <td>00000000000000000000000000000000</td> 
  </tr> 
  <tr> 
   <td>STOP</td> 
   <td>00000000000000000000000000000001</td> 
  </tr> 
  <tr> 
   <td>TIDYING</td> 
   <td>00000000000000000000000000000010</td> 
  </tr> 
  <tr> 
   <td>TERMINATED</td> 
   <td>00000000000000000000000000000011</td> 
  </tr> 
 </tbody> 
</table>

**左移之后的状态值，注意观察高三位，通过后续分析发现，高三位用来表示线程池的状态值，后29位用来表示线程数。**

<table> 
 <thead> 
  <tr> 
   <th>状态值</th> 
   <th>二进制</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>RUNNING</td> 
   <td>11100000000000000000000000000000</td> 
  </tr> 
  <tr> 
   <td>SHUTDOWN</td> 
   <td>00000000000000000000000000000000</td> 
  </tr> 
  <tr> 
   <td>STOP</td> 
   <td>00100000000000000000000000000000</td> 
  </tr> 
  <tr> 
   <td>TIDYING</td> 
   <td>01000000000000000000000000000000</td> 
  </tr> 
  <tr> 
   <td>TERMINATED</td> 
   <td>01100000000000000000000000000000</td> 
  </tr> 
 </tbody> 
</table>

private static final int CAPACITY   = (1 << COUNT_BITS) - 1;//后29位最大值。

**得到线程池的状态计算函数：**

private static int runStateOf(int c)     {  return c & ~CAPACITY; }

![在这里插入图片描述][20200511001852648.png]  
**得到线程池的数量计算函数：**

private static int workerCountOf(int c)  {  return c & CAPACITY; }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDkyNDYy_size_16_color_FFFFFF_t_70]  
**线程池五种状态说明：**

**1. RUNNING**  
状态说明：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。  
状态切换：线程池的初始化状态是RUNNING。换句话说，线程池被一旦被创建，就处于RUNNING状态，并且线程池中的任务数为0！

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

ctlof（）函数计算出线程池数量。  
**2. SHUTDOWN**  
状态说明：线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。  
状态切换：调用线程池的shutdown()接口时，线程池由RUNNING -> SHUTDOWN。  
**3. STOP**  
(1) 状态说明：线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。  
(2) 状态切换：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。  
**4. TIDYING**  
(1) 状态说明：当所有的任务已终止，ctl记录的”任务数量(workercount)”为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。  
(2) 状态切换：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。  
当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。  
**5. TERMINATED**  
(1) 状态说明：线程池彻底终止，就变成TERMINATED状态。  
(2) 状态切换：线程池处在TIDYING状态时，执行完terminated()之后，就会由 TIDYING -> TERMINATED。  
**线程池的构造函数，一共有四个构造函数，最终调用的都是一个：**

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;
        }

各个参数解释如下：

*  corePoolSize: 核心线程数量，可以类比正式员工数量，常驻线程数量。
 *  maximumPoolSize: 最大的线程数量，公司最多雇佣员工数量。常驻+临时线程数量。
 *  workQueue：多余任务等待队列，再多的人都处理不过来了，需要等着，在这个地方等。
 *  keepAliveTime：非核心线程空闲时间，就是外包人员等了多久，如果还没有活干，解雇了。
 *  threadFactory: 创建线程的工厂，在这个地方可以统一处理创建的线程的属性。每个公司对员工的要求不一样，恩，在这里设置员工的属性。
 *  handler：线程池拒绝策略，什么意思呢?就是当任务实在是太多，人也不够，需求池也排满了，还有任务咋办?默认是不处理，抛出异常告诉任务提交者，我这忙不过来了。  
    **接下来我们来看看往线程池添加一个线程的方法：**

public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
            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);
        }

workerCountOf()函数计算线程池当前有多少线程数，前面有提到过，addWorker(command,true)该函数两个参数，一个用来添加的线程，另一个标志是否创建核心线程还是普通线程数。isRuning()该函数主要判断线程池是否正在运行。  
该函数流程图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDkyNDYy_size_16_color_FFFFFF_t_70 1]  
**接下来看看添加一个线程：**

private boolean addWorker(Runnable firstTask, boolean core) { 
            retry:
            for (;;) { 
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
    
                for (;;) { 
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try { 
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) { 
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try { 
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int rs = runStateOf(ctl.get());
    
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                            workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally { 
                        mainLock.unlock();
                    }
                    if (workerAdded) { 
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

前面两个for循环判断，线程池的状态跟容量是否能添加新的线程，不能返回false。  
后面try开始，首先加锁，判断线程池的状态，然后添加线程到线程池里，最后启动线程。

接下来我们看看workers:

private final HashSet<Worker> workers = new HashSet<Worker>();

通过这行代码我们发现，线程池的底层结构是一个HashSet，而HashSet又是Hashmap的实现者。  
**最后我们来看看线程运行：**

final void runWorker(Worker w) { 
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try { 
                while (task != null || (task = getTask()) != null) { 
                    w.lock();
                    // If pool is stopping, ensure thread is interrupted;
                    // if not, ensure thread is not interrupted. This
                    // requires a recheck in second case to deal with
                    // shutdownNow race while clearing interrupt
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try { 
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try { 
                            task.run();
                        } catch (RuntimeException x) { 
                            thrown = x; throw x;
                        } catch (Error x) { 
                            thrown = x; throw x;
                        } catch (Throwable x) { 
                            thrown = x; throw new Error(x);
                        } finally { 
                            afterExecute(task, thrown);
                        }
                    } finally { 
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
                processWorkerExit(w, completedAbruptly);
            }
        }

getTask()去队列中取出线程。  
这里有两个钩子(beforeExecute，afterExecute)允许我们自己继承线程池，做任务执行前后处理。  
有些地方说得不太准确，请大家批评指正。  
参考文献：https://www.cnblogs.com/rinack/p/9888717.html?tdsourcetag=s\_pctim\_aiomsg

[20200511001852648.png]: /images/20230312/ae78a14e70904d71ad3387b7ec0e5ed9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDkyNDYy_size_16_color_FFFFFF_t_70]: /images/20230312/699bde24533d4f6193dec1fd78b0626d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NDkyNDYy_size_16_color_FFFFFF_t_70 1]: /images/20230312/604a9353b66d49ae9aeebf2626c169ab.png