java 线程池源码解析

川长思鸟来 2021-09-23 17:46 433阅读 0赞

jdk版本：1.8

java中的线程池通常指的是ThreadPoolExecutor，本文主要解析线程池执行过程中的核心代码。

### 文章目录 ###

*   *   *  为什么使用线程池
         *  线程池的核心思想
         *  继承关系
         *  ThreadPoolExecutor的主要成员变量
         *  构造方法
         *   *  Executors
         *  源码解析
         *   *  execute方法
             *   *  核心线程与非核心线程
             *  addWorker
             *  runWorker

### 为什么使用线程池 ###

先来说一下为什么使用线程池吧。我们平常所写的那些java代码（不含Thread，Runnable的）都是单线程的，比如这样

public static void main(String[] args) throws InterruptedException { 
    		// TODO Auto-generated method stub
    		for(int i = 0; i < 5; i++) { 
    			System.out.println(i);
    			Thread.sleep(500);//do something
    		}
    	}

但是这种写法在执行效率上是存在瓶颈的，上面的代码中，1秒钟处理2条数据，10秒钟也只能处理20条。于是，就要用到多线程的写法。

public static void main(String[] args) { 
    		// TODO Auto-generated method stub
    		for (int i = 0; i < 5; i++) { 
    			new Thread(new Runnable() { 
    				public void run() { 
    					for (int i = 0; i < 5; i++) { 
    						System.out.println(i);
    						try { 
    							Thread.sleep(500);// do something
    						} catch (InterruptedException e) { 
    							// TODO Auto-generated catch block
    							e.printStackTrace();
    						}
    					}
    				}
    			}).start();
    		}
    
    	}

我在上面的代码中开启了5个线程，因此获得了5倍的处理效率。所以，多线程是用来提升处理效率的（多线程提升效率的本质在于更高效的利用了cpu，只有在合适的场景下才会提高效率。在某些特定情况下，开启多线程甚至可能会降低效率）；

但是，上面的多线程写法是存在许多问题的，比如，java中的线程属于稀缺资源，频繁的new线程会大量消耗系统资源；无法对线程的数量进行比较好的控制；无法对线程的执行进行控制；无法监控线程的执行状态等等。虽然所有这些都可以通过写额外的代码来实现，但其实jdk本身已经做了一整台完善的封装，这就是ThreadPoolExecutor。

总结来说，ThreadPoolExecutor就是针对多线程处理中，可能出现的各种情况所封装好的一套工具类。

### 线程池的核心思想 ###

在解析源码之前，先来说一下线程池的核心思想吧。在多线程处理中，最消耗资源的当属新建线程了，线程池所做的核心工作就是降低创建线程的操作。先来看一下线程池的简单示例吧

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Test { 
    	
    	public static int cot = 0;
    	/** * @param args * @throws InterruptedException */
    	public static void main(String[] args) throws InterruptedException { 
    		// TODO Auto-generated method stub
    		
    		ThreadPoolExecutor executor = 
    				new ThreadPoolExecutor(5, 5, 10000, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(10));
    		MyTask myTask = new MyTask(1);
    		for(int i = 0; i < 10; i++){ 
    			executor.execute(myTask);
    			System.out.println("pool size = " + executor.getPoolSize() + " queue size = " + executor.getQueue().size() + 
    					"completed size = " + executor.getCompletedTaskCount());
    		}
    		
    		executor.shutdown();
    	}
    
    }
    
    
    class MyTask implements Runnable{ 
    	
    	private int taskNum;
    	
    	public MyTask(int taskNum) { 
    		// TODO Auto-generated constructor stub
    		this.taskNum = taskNum;
    	}
    
    	@Override
    	public void run() { 
    		// TODO Auto-generated method stub
    		int temp = Test.cot++;
    		System.out.println("start task " + temp + " start");
    		
    		try { 
    			Thread.currentThread().sleep(1000);
    		} catch (InterruptedException e) { 
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    			System.out.println("catch");
    		}
    		
    		System.out.println("end task " + temp + " end");
    		
    	}
    	
    }

运行之后，观察控制台可以发现，它是以每五个一组运行的，并且也只有5个线程在执行这10个任务。这也是线程池的意义所在----**创建过的线程在使用后不进行销毁，而是暂时存起来，并在合适的时候继续执行的任务。**

想要理解线程池的源码，最好是要先了解这种使线程重复利用的原理，因为整个线程池就是以此为基础进行扩展的。而这个核心代码其实相当简单，完全可以自己写一个来加深理解。下面是我从源码中抄的一部分加上一些简化后写出的简易代码。

import java.util.Stack;
    
    public class TdTest01 { 
    
    	public static void main(String[] args) throws InterruptedException { 
    		Runnable r1 = new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println(111);
    			}
    		};
    		
    		MyThread td = new MyThread(r1);
    		
    		td.execute(r1);
    		td.execute(r1);
    		
    		Thread.sleep(1000);
    		
    		td.execute(r1);
    		td.execute(r1);
    	}
    
    }
    
    class MyThread implements Runnable{ 
    
    	Runnable target;//其实只有初始化时用到了
    	
    	Stack<Runnable> queue = new Stack<Runnable>();//无界缓存队列
    
    	public MyThread(Runnable target) { 
    		super();
    		this.target = target;
    		new Thread(this).start();//初始化
    	}
    	//对应源码的execute方法
    	public void execute(Runnable target) { 
    		//源码中此处有大量针对状态的判断
    		this.queue.add(target);
    	}
    	//死循环来从缓存队列中获取任务
    	private Runnable getTask() { 
    		for(;;) { 
    			synchronized (queue) { 
    				if(queue.size() > 0 && queue.peek() != null) { 
    					return queue.pop();
    				}
    			}
    			//源码中会在此处判断线程池状态，若以关闭，则返回空，进而使下面的while循环终止，并进入收尾工作
    		}
    	}
    
    	@Override
    	public void run() { 
    		Runnable task = target;
    		target = null;
    		while(task != null || (task = getTask()) != null) { 
    			task.run();//实际所执行的代码
    			task = null;
    		}
    	}
    	
    }

上面代码中的`MyThread`即是一个可重复利用的线程（实际上线程池中会维护多个`MyThread`，但是与说明内容无关就省略了），原理非常简单，就是一个死循环的线程，其中会不断的尝试从缓存队列中获取任务来执行。  
jdk中的线程池（ThreadPoolExecutor）就是以此为基础进行扩展的，简单来说，主要扩展的就是，

*  将这种可重复利用的线程以集合的形式维护起来
 *  对缓存队列添加了各种控制方法
 *  对线程池加入了状态，如running，shutdown等，并针对状态做了大量判断

其中最麻烦的是最后一个，多线程中的状态判断如果自己写的话是非常容易出错的，在源码中会看到大量双重判断的写法，如果基本功不足，理解起来会很累。

### 继承关系 ###

Executor  
—\[extends\]—  
ExecutorService  
—\[implements\]—  
AbstractExecutorService  
—\[extends\]—  
ThreadPoolExecutor

Executor是根接口，只定义了一个方法`void execute(Runnable command);`\------执行一个线程。线程所需要做的也正是执行任务。

ExecutorService是继承了Executor的一个接口，主要定义了添加了3套方法

*  关于状态的方法，比如修改状态和获取状态（shutdown、isShutdown、isTerminated等等）
 *  submit，执行方法并返回一个Future对象-------可以用来获取结果。算是对execute方法的扩展了
 *  各种invoke方法-------以集合为参数，并进行执行，实现上通常是对集合中的对象调用submit方法

关于状态的方法很常用，但是submit、invoke则很少见，甚至在ThreadPoolExecutor中都很难找到相关的调用。

AbstractExecutorService是一个抽象类，对ExecutorService进行了基础的实现。

ThreadPoolExecutor是实际使用的线程类。

### ThreadPoolExecutor的主要成员变量 ###

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private final HashSet<Worker> workers = new HashSet<Worker>();
    private final BlockingQueue<Runnable> workQueue;
    
    private int largestPoolSize;
    private long completedTaskCount;
    
    private volatile RejectedExecutionHandler handler;
    
    private volatile long keepAliveTime;
    private volatile boolean allowCoreThreadTimeOut;
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;

`ctl` 是用一个数字来记录线程池的状态与工作线程的数量，数字的前面记录状态，后面是线程数。  
`workers` 是工作线程的集合。  
`workQueue`是缓存队列。  
`largestPoolSize` 和`completedTaskCount` 是统计用数据，表示池中曾经最大的线程数和执行成功的线程。  
`handler` 是拒绝策略，主要是在缓存队列满时调用。  
后面4个是初始化时配上的参数，用来表示超时时间与线程容量。

### 构造方法 ###

简单的调用例子

new ThreadPoolExecutor(5, 5, 10000, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(10));

所执行的构造方法

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) { 
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }
    
    	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;
        }

就是将参数设置上，并尽可能的做了些异常判断。

#### Executors ####

顺带提一下`Executors`，这是一个帮助建立线程池的工具类，本质上是一个简单工厂，里面都是类似`new ThreadPoolExecutor(5, 5, 10000, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(10));` 这样的方法，通过参数的控制来达到某种效果。比如

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { 
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>(),
                                        threadFactory));
        }

将corePoolSize 、maximumPoolSize 同时设为1，就成了单线程运行的线程池------单例线程池。

### 源码解析 ###

线程池的运行主要由三个方法来支持：execute、addWorker、runWorker，下面依次讲解。

#### execute方法 ####

源码

/** * Executes the given task sometime in the future. The task * may execute in a new thread or in an existing pooled thread. * * If the task cannot be submitted for execution, either because this * executor has been shutdown or because its capacity has been reached, * the task is handled by the current {@code RejectedExecutionHandler}. * * @param command the task to execute * @throws RejectedExecutionException at discretion of * {@code RejectedExecutionHandler}, if the task * cannot be accepted for execution * @throws NullPointerException if {@code command} is null */
        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);
        }

这是所有的方法的入库，使用线程池来执行任务只需要调用这一个方法即可。代码很少，主要就是判断状态、创建工作线程、向缓存队列中添加任务。

把注释用谷歌翻译一遍

> 在将来的某个时间执行给定的任务（若工作线程已经排满，将不会立刻执行，若空闲，则会自动去执行）。 该任务可以在新线程或现有池线程中执行（没有工作线程，比如刚初始化的时候，会创建一个新线程并去执行，否则会使用旧线程执行）。 如果由于该执行器已关闭或已达到其能力而无法提交执行任务，则该任务由当前的RejectedExecutionHandler处理（RejectedExecutionHandler就是阻塞策略，会在线程关闭或者缓存队列满时调用）。  
>   
> 1.如果正在运行的线程少于corePoolSize线程，会尝试执行以下操作：  
> 使用给定的命令作为第一个启动新线程  
> 任务。 对addWorker的调用自动检查runState和  
> workerCount，因此可以防止假警报的增加  
> 通过返回false返回不应该执行的线程。（就是对状态的双重校验，确保准确性）  
>   
> 2.如果任务可以成功排队（添加到缓存队列），那么我们仍然需要  
> 仔细检查我们是否应该添加线程  
> （因为现有的自上次检查后死亡）或  
> 自从进入此方法以来，该池已关闭。 所以我们  
> 重新检查状态，并在必要时回退排队  
> 停止，或者如果没有线程，则启动一个新线程。  
>   
> 3.如果我们无法将任务排队，那么我们尝试添加一个新的  
> 线（对于代码中的elseif，实际添加的是非核心线程）。 如果失败，我们知道我们已经关闭或饱和  
> 因此拒绝任务。

逐句翻译一下

int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) { 
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }

获取状态码，如果当前线程数小于核心线程，则创建核心线程并return，否则向下执行，以及重新获取状态码，用以防止状态在addWorker的时候被串改（双重校验）。

//如果线程池运行状态为running，并且新任务加入到工作队列中（在正常情况下，此时新任务已经进入到准备执行阶段，即将被工作线程调用）
    		if (isRunning(c) && workQueue.offer(command)) { 
                int recheck = ctl.get();//获取线程池状态
                //如果线程池不在running状态，则移除任务
                if (! isRunning(recheck) && remove(command))
                    reject(command);//执行拒绝策略
                //如果当前没有工作线程
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);//新建一个工作线程（这个工作线程会自己到缓存队列中获取任务执行）
            }
            else if (!addWorker(command, false))//无法添加新工作线程（原因后面说）
                reject(command);//执行拒绝策略

（上面就是execute方法的全部了，很简单，因为麻烦的都在addWorker里。）

通过这一段代码，可以很准确的理解核心线程、非核心线程、缓存池之间的关系了。  
假设线程池是这样创建的

ThreadPoolExecutor executor = 
    				new ThreadPoolExecutor(1, 2, 10000, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(3));

那么，其核心线程数为1，最大线程数为2，非核心线程数为2 - 1 = 1 。缓存队列为3 。在连续接收到5个任务时，它的执行顺序是这样的

1.  接收到第一个任务，创建一个核心线程并立即执行。
2.  接收到第二个任务，创建一个非核心线程并立即执行。
3.  又接收到3个任务，由于工作线程达到最大值，会将3个任务放入缓存队列
4.  如果之后又立刻收到任务，会执行拒绝策略
5.  最终，所有任务会以两个为一组进行执行。

写段代码测试一下

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Test { 
    	
    	public static int cot = 0;
    	/** * @param args * @throws InterruptedException */
    	public static void main(String[] args) throws InterruptedException { 
    		// TODO Auto-generated method stub
    		ThreadPoolExecutor executor = 
    				new ThreadPoolExecutor(1, 2, 10000, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(3));
    		MyTask myTask = new MyTask(1);
    		for(int i = 0; i < 5; i++){ 
    			executor.execute(myTask);
    			System.out.println("pool size = " + executor.getPoolSize() + " queue size = " + executor.getQueue().size() + 
    					" completed size = " + executor.getCompletedTaskCount());
    		}
    		
    		Thread.sleep(5000);
    		
    		System.out.println();
    		System.out.println();
    		System.out.println();
    		
    		for(int i = 0; i < 5; i++){ 
    			executor.execute(myTask);
    			System.out.println("pool size = " + executor.getPoolSize() + " queue size = " + executor.getQueue().size() + 
    					"completed size = " + executor.getCompletedTaskCount() + " time = " + System.currentTimeMillis());
    		}
    		
    		executor.shutdown();
    	}
    
    }
    
    
    class MyTask implements Runnable{ 
    	
    	private int taskNum;
    	
    	public MyTask(int taskNum) { 
    		// TODO Auto-generated constructor stub
    		this.taskNum = taskNum;
    	}
    
    	@Override
    	public void run() { 
    		// TODO Auto-generated method stub
    		int temp = Test.cot++;
    		System.out.println("start task " + temp + " start" + " time = " + System.currentTimeMillis());
    		
    		try { 
    			Thread.currentThread().sleep(1000);
    		} catch (InterruptedException e) { 
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    			System.out.println("catch");
    		}
    		
    		System.out.println("end task " + temp + " end");
    		
    	}
    	
    }

控制台输出

pool size = 1 queue size = 0 completed size = 0
    pool size = 1 queue size = 1 completed size = 0
    pool size = 1 queue size = 2 completed size = 0
    pool size = 1 queue size = 3 completed size = 0
    pool size = 2 queue size = 3 completed size = 0
    start task 0 start time = 1579501981502
    start task 1 start time = 1579501981506
    end task 0 end
    start task 2 start time = 1579501982502
    end task 1 end
    start task 3 start time = 1579501982506
    end task 2 end
    start task 4 start time = 1579501983502
    end task 3 end
    end task 4 end
    
    
    
    start task 5 start time = 1579501986502
    pool size = 1 queue size = 1completed size = 5 time = 1579501986502
    pool size = 1 queue size = 1completed size = 5 time = 1579501986503
    pool size = 1 queue size = 2completed size = 5 time = 1579501986503
    pool size = 1 queue size = 3completed size = 5 time = 1579501986503
    pool size = 2 queue size = 3completed size = 5 time = 1579501986503
    start task 6 start time = 1579501986504
    end task 5 end
    start task 7 start time = 1579501987502
    end task 6 end
    start task 8 start time = 1579501987504
    end task 7 end
    start task 9 start time = 1579501988502
    end task 8 end
    end task 9 end

##### 核心线程与非核心线程 #####

这里要额外说一下核心线程与非核心线程，这名字实在有点唬人，看起来区别很大，但在实际运行中，却几乎没什么区别。这是因为在工作线程类`ThreadPoolExecutor.Worker`中根本就没有什么是否为核心的属性，这个核心线程数量`corePoolSize` 基本只在创建线程的时候才会使用到。  
非要说核心线程与非核心线程的区别的话，就是：**线程池首先会尝试只使用核心线程（更准确的说应该是，核心线程数量即`corePoolSize`数量的线程）来处理任务，如果任务太多了，则会将多余的任务放到缓存队列中，如果连缓存队列都放满了，才会创建非核心线程（更准确的说应该是，创建最大线程数量即`maximumPoolSize`数量的线程）来处理任务。  
说得更简单一点，就是，因为线程是稀缺资源，所以，线程池将会尝试只创建`corePoolSize`数量的线程来处理任务，如果不能，才会创建`maximumPoolSize`数量的线程来处理任务。**

#### addWorker ####

这个方法在execute中多次出现，先看源码

/* * Methods for creating, running and cleaning up after workers */
    
        /** * Checks if a new worker can be added with respect to current * pool state and the given bound (either core or maximum). If so, * the worker count is adjusted accordingly, and, if possible, a * new worker is created and started, running firstTask as its * first task. This method returns false if the pool is stopped or * eligible to shut down. It also returns false if the thread * factory fails to create a thread when asked. If the thread * creation fails, either due to the thread factory returning * null, or due to an exception (typically OutOfMemoryError in * Thread.start()), we roll back cleanly. * * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */
        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;
        }

最上面的注释很好的概括了这个方法做了什么

> Methods for creating, running and cleaning up after workers

先说一下参数

*  firstTask：在创建新工作线程后立即执行的任务，若为空，则相当于单纯的创建一个工作线程（之后会自动从缓存队列中获取任务）。
 *  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
                }
            }

这是一个双重for循环，主要是判断了线程池当前状态是否允许创建新的线程，并做了一个计数。

逐行来看

int c = ctl.get();
                int rs = runStateOf(c);

获取线程池状态

if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;

（这段代码差不多是所有源码中最烦的，一堆逻辑判断）

要看懂这一段，首先要知道线程池中的运行状态是以数字形式存储的，从小到大依次为  
**RUNNING，SHUTDOWN ，STOP，TIDYING ，TERMINATED**  
RUNNING为正常运行状态；SHUTDOWN为不再处理新任务，但是会处理队列中的任务；后面3个现在可以简单的认为是终止状态。

所以这段代码翻译一下就是：  
**当线程池不在RUNNING状态时，并且线程池状态为SHUTDOWN ，并且firstTask（创建线程后立即执行的任务）为空，并且当前缓存队列不为空时，允许创建新线程。**

这个判断的作用是完美对应**SHUTDOWN**状态的含义的，用人类的语言再来对这段代码解释一下：***当状态为SHUTDOWN时，允许创建新线程来处理缓存队列中堆积的任务。***

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
                }

这段代码只有这一段需要解释一下 `wc >= (core ? corePoolSize : maximumPoolSize))` ，这是判断是否超过核心（最大）线程数量的判断。剩下的就是计数和双重判断。

第二部分

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());
    					//判断线程池状态，为RUNNING或者为SHUTDOWN中并且创建空工作线程时往下执行
                        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;//largestPoolSize 计数，统计线程池中曾经达到过的最大线程数
                            workerAdded = true;
                        }
                    } finally { 
                        mainLock.unlock();
                    }
                    if (workerAdded) { 
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;

这段代码很简单，解释写到注释里。

#### runWorker ####

在上一节中有这样一段代码

if (workerAdded) { 
                        t.start();
                        workerStarted = true;
                    }

它实际上会调用`Worker`类的run方法

/** Delegates main run loop to outer runWorker */
            public void run() { 
                runWorker(this);
            }

runWorker的源码

/** * Main worker run loop. Repeatedly gets tasks from queue and * executes them, while coping with a number of issues: * * 1. We may start out with an initial task, in which case we * don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and then we * ensure that unless pool is stopping, this thread does not have * its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to afterExecute. * We separately handle RuntimeException, Error (both of which the * specs guarantee that we trap) and arbitrary Throwables. * Because we cannot rethrow Throwables within Runnable.run, we * wrap them within Errors on the way out (to the thread's * UncaughtExceptionHandler). Any thrown exception also * conservatively causes thread to die. * * 5. After task.run completes, we call afterExecute, which may * also throw an exception, which will also cause thread to * die. According to JLS Sec 14.20, this exception is the one that * will be in effect even if task.run throws. * * The net effect of the exception mechanics is that afterExecute * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */
        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);
            }
        }

这段代码与我在一开始写的`MyThread` 的作用基本是一致的，只是要更完善一些。  
代码太多了，简化一下再来看

Runnable task = w.firstTask;
    		w.firstTask= null;
    		//getTask中有一个死循环，在不断的从缓存队列中获取任务
    		while(task != null || (task = getTask()) != null) { 
    			//状态判断
    			beforeExecute(wt, task);//相当于aop了，在执行任务前执行
    			task.run();//实际所执行的代码
    			afterExecute(task, thrown);//在执行任务后执行
    			task = null;
    		}
    		processWorkerExit(w, completedAbruptly);//退出时调用，做一些计数和移除的工作