浅谈java线程池实现

2019-04-28 08:22:05来源：博客园阅读 ()

再进入主题之前，我们先了解几个概念，对读源码有所帮助，对于线程池的运行状态，有4个级别，分别是RUNNING，SHUTING，STOP，TIDING，TERMINATED
解释如下：

The runState provides the main lifecycle control, taking on values:
     *
     *   RUNNING:  Accept new tasks and process queued tasks            //能接受新的任务，并且可以运行已经在任务队列中的任务
     *   SHUTDOWN: Don't accept new tasks, but process queued tasks     //不再接受新的任务，但是仍能运行在阻塞队列里等待的任务
     *   STOP:     Don't accept new tasks, don't process queued tasks,  //不再接受新的任务，也停止运行队列中的任务
     *             and interrupt in-progress tasks
     *   TIDYING:  All tasks have terminated, workerCount is zero,      //后面的就是停止后的东西了，暂时不需要理解。。
     *             the thread transitioning to state TIDYING
     *             will run the terminated() hook method
     *   TERMINATED: terminated() has completed

　将这几个任务对应着数字，所以可以进行大小比较

1     private static final int RUNNING    = -1 << COUNT_BITS;
2     private static final int SHUTDOWN   =  0 << COUNT_BITS;
3     private static final int STOP       =  1 << COUNT_BITS;
4     private static final int TIDYING    =  2 << COUNT_BITS;
5     private static final int TERMINATED =  3 << COUNT_BITS;

下面我们正式了解线程池：

线程池的实现有很多，比如

    1. newFixedThreadPool()        //可以指定核心线程个数，并且创建的所有线程都是核心线程
	2. newSingleThreadExecutor()   //有且只有一个线程，且是核心线程
	3. newCachedThreadPool()       //没有规定最多可以有多少个线程，但没有一个是核心线程
	4. newScheduledThreadPool()		//可以指定核心线程的个数，而且可以创建非核心线程（不限数量）

这些线程的创建方法其实最终都引用了一个构造方法：

1 public ThreadPoolExecutor(
2           int corePoolSize,     //核心线程个数
3           int maximumPoolSize,   //最大线程个数
4           long keepAliveTime,     //线程空闲多长时间被回收
5           TimeUnit unit,         //时间的单位
6           BlockingQueue<Runnable> workQueue,   //任务队列
7           ThreadFactory threadFactory,       //线程工厂方法（其实就是给线程设置一下属性，可以用它创建线程）
8           RejectedExecutionHandler handler)  //回绝策略

　　这其中的线程工厂方法和回绝策略可以自己创建，也可以使用默认的策略，回绝策略我会在待会儿说明

现在我们的关注点就在ThreadPoolExecutor对象上了，因为他负责管理所有线程池里的线程，以及维护阻塞队列
那么他是怎么维护线程和任务的呢？我们来看一下这个类的结构

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

它里面有一个Worker类型的hashset，以及一个任务类型的BlockingQueue，
所以我们可以猜到，这个Hashset的workers就是用来存储线程的，而BlockingQueue就是用来存储任务的
我们再仔细了解一下这个Worker：

 1 private final class Worker
 2         extends AbstractQueuedSynchronizer
 3         implements Runnable
 4     {
 5         
 6         final Thread thread;
 7       
 8         Runnable firstTask;                           //创建worker时给的初始任务
 9        
10         volatile long completedTasks;
11 
12       
13         Worker(Runnable firstTask) {
14             setState(-1); 
15             this.firstTask = firstTask;                          //创建worker时给worker的初始任务
16             this.thread = getThreadFactory().newThread(this);    //利用线程工厂创建一个新的线程
17         }
18 
19         public void run() {
20             runWorker(this);
21         }
22     }

这个Worker里面包含着一个Thread，并且Worker本身也是一个任务（继承了Runnable），为什么还要传给他一个初始任务呢？为什么要这么设计呢？
答案其实很简单，我们都知道一个Thread不能被重用的，但是跟据线程池的定义，一个线程却可以在完成一个任务之后又去做其他的任务，这就是源于这个设计，这个Thread一直运行的就是worker这个任务，而没有去运行其他的任务，而worker的run方法里面有一个runworker方法，线程就是通过这个runworker方法去读取初始任务或者任务列表中的任务来运行的

有了这些基础知识后，我们就开始从我们常用的线程流程开始分析，看看线程到底是怎么运作的吧！

我们都知道，运行线程池时要给他提交任务，比如

ExecutorService exec = Executors.newCachedThreadPool(); //创建线程池
exec.execute(Runnable) //提交自己的任务

我们跟踪一下execute方法，他的描述如下

 1    public void execute(Runnable command) {
 2         if (command == null)
 3             throw new NullPointerException();
 4         
 5          //获取线程池状态码
 6         int c = ctl.get();   
 7         
 8         //workerCountof(c) 能根据状态码运算出线程池核心线程的个数 
 9         if (workerCountOf(c) < corePoolSize) {   
10             if (addWorker(command, true))    
11                 return;
12             c = ctl.get();    
13         }
14         //如果核心线程已经满了，就尝试添加任务到队列
15         if (isRunning(c) && workQueue.offer(command)) {    
16             int recheck = ctl.get();       
17             //如果添加成功，但此时发现线程池已经停止接受任务，则删除任务            
18             if (! isRunning(recheck) && remove(command))   
19                 reject(command);
20             
21             //如果线程池没有停止接受，但是核心线程已经全部被回收(比如cachedThreadpool会发生这种情况)
22             else if (workerCountOf(recheck) == 0)     
23                 
24             //添加非核心线程，且这个线程没有初始任务（会让他在任务队列中去取）
25                 addWorker(null, false);                    
26                                                          
27             //这里隐含了一层意思：这些条件都不满足时，即有核心线程，且队列未满，则什么都不做，仅仅将任务添加到队列
28         }
29         //如果核心线程满了，并且任务队列也添加失败了（满了），那么尝试创建非核心线程
30         else if (!addWorker(command, false)) 
31             
32             //如果都失败了，那么启用拒绝策略
33             reject(command);                              
34     }

跟据上面的方法，我们知道execute主要就是判断线程池情况，然后决定是直接新建线程来执行任务，或者添加任务到任务列表，其最主要的方法就是addWorker
所以我们再来看一下addWorker是怎么实现的

 1 private boolean addWorker(Runnable firstTask, boolean core) {
 2         retry:
 3         /***********增加线程池的线程计数*************************/
 4         for (;;) {
 5             int c = ctl.get();
 6             int rs = runStateOf(c);    //获取当前运行状态RunStatus
 7             
 8             if (rs >= SHUTDOWN &&
 9                 ! (rs == SHUTDOWN &&   //如果已经停止
10                    firstTask == null &&  //或者任务非空
11                    ! workQueue.isEmpty()))  //或者工作队列为空
12                 return false;
13 
14             for (;;) {
15                 int wc = workerCountOf(c);
16                 if (wc >= CAPACITY ||
17                     wc >= (core ? corePoolSize : maximumPoolSize)) //判断是否大于规定线程数
18                     return false;
19                 if (compareAndIncrementWorkerCount(c))     //符合条件则增加线程池的线程计数
20                     break retry;
21                 c = ctl.get();  // Re-read ctl
22                 if (runStateOf(c) != rs)
23                     continue retry;
24                 // else CAS failed due to workerCount change; retry inner loop
25             }
26         }
27     /************创建新线程，并将它加入wokers的hashSet进行管理********************/
28         boolean workerStarted = false;
29         boolean workerAdded = false;
30         Worker w = null;
31         try {
32             w = new Worker(firstTask);
33             final Thread t = w.thread;
34             if (t != null) {
35                 final ReentrantLock mainLock = this.mainLock;
36                 mainLock.lock();
37                 try {
38                     // Recheck while holding lock.
39                     // Back out on ThreadFactory failure or if
40                     // shut down before lock acquired.
41                     int rs = runStateOf(ctl.get());
42 
43                     if (rs < SHUTDOWN ||
44                         (rs == SHUTDOWN && firstTask == null)) {
45                         if (t.isAlive()) // precheck that t is startable
46                             throw new IllegalThreadStateException();
47                         workers.add(w);
48                         int s = workers.size();
49                         if (s > largestPoolSize)
50                             largestPoolSize = s;
51                         workerAdded = true;
52                     }
53                 } finally {
54                     mainLock.unlock();
55                 }
56                 if (workerAdded) {
57                     t.start();  //运行线程，对应着worker的run方法，里面调用了runworker（this）方法
58                     workerStarted = true;
59                 }
60             }
61         } finally {
62             if (! workerStarted)
63                 addWorkerFailed(w);
64         }
65         return workerStarted;
66     }

　　由于线程运行后，run方法里调用的是runworker（this）方法,所以我们再来看一看这个方法

 1  1 final void runWorker(Worker w) {
 2  2         Thread wt = Thread.currentThread();
 3  3         Runnable task = w.firstTask;
 4  4         w.firstTask = null;     //运行一次初始任务后，就应该将初始任务放掉，否则会重复运行
 5  5         w.unlock(); // allow interrupts
 6  6         boolean completedAbruptly = true;
 7  7         try {
 8  8             while (task != null || (task = getTask()) != null) {    
 9  9                 w.lock();       
10 10                 if ((runStateAtLeast(ctl.get(), STOP) ||
11 11                      (Thread.interrupted() &&
12 12                       runStateAtLeast(ctl.get(), STOP))) &&
13 13                     !wt.isInterrupted())
14 14                     wt.interrupt();
15 15                 try {
16 16                     beforeExecute(wt, task);
17 17                     Throwable thrown = null;
18 18                     try {
19 19                         task.run();            //运行任务，不是用thread.start,而是直接调用任务的run方法
20 20                     } catch (RuntimeException x) {
21 21                         thrown = x; throw x;
22 22                     } catch (Error x) {
23 23                         thrown = x; throw x;
24 24                     } catch (Throwable x) {
25 25                         thrown = x; throw new Error(x);
26 26                     } finally {
27 27                         afterExecute(task, thrown);
28 28                     }
29 29                 } finally {
30 30                     task = null;                    
31 31                     w.completedTasks++;
32 32                     w.unlock();
33 33                 }
34 34             }
35 35             completedAbruptly = false;        //没有任务了，这个线程空闲出来了
36 36         } finally {
37 37             processWorkerExit(w, completedAbruptly); //将线程从hashset里面移除
38 38         }
39 39     }

那么到现在我们已经大致了解了，线程池的运行流程，总结一下就是：

1.创建线程池后，首先会用submit或者execute提交任务
2.任务提交后，会根据线程池的状态利用AddWorker进行线程的创建，并将其添加到线程池中，判断条件如下：
　　线程数量未达到corePoolSize，则新建一个线程(核心线程)执行任务
　　线程数量达到了corePoolsize，则将任务移入队列等待
　　队列已满，新建线程(非核心线程)执行任务
　　队列已满，总线程数又达到了maximumPoolSize，就会采用回绝策略进行处理
3.如果线程添加成功，此时会调用线程的start方法，而线程对应任务的run方法里面，调用的是runworker（）方法
4.runworker()方法会在线程的初始任务或者任务队列里面取任务来运行，运行时直接调用被取出任务的run方法
5.运行完成后，如果不能取得新的任务，那么此时线程就是空闲线程，会执行清理策略（跟据空闲时间的长短来清理）

现在问题已经被我们解决了，但是还有一个问题，细心的同学们应该已经发现了：不是说好的线程池里的核心线程可以不被回收吗，但是按照源码的分析，
线程池里所有的线程，只要拿不到任务，过一段时间就会被回收，这是怎么回事呢？其实原因在于我们忽略掉的getTask方法，我们来看看它的源码

 1 private Runnable getTask() {
 2         boolean timedOut = false; // Did the last poll() time out?
 3 
 4         for (;;) {
 5             int c = ctl.get();
 6             int rs = runStateOf(c);
 7 
 8             // Check if queue empty only if necessary.
 9             if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { //如果线程池已经停止接受新线程，且工作队列为空
10                 decrementWorkerCount();                       //释放线程
11                 return null;                                  //返回空（使线程拿不到任务）
12             }
13 
14             int wc = workerCountOf(c);
15 
16            //或操作有false才往后判断
17             //allowCoreThreadTimeOut表示核心线程也可以回收
18             boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;  
19         //当允许回收core线程，timed为true
20         //当不允许回收核心线程时，如果当前线程数> 核心线程数   timed=true  否则为false
21 
22             if ((wc > maximumPoolSize || (timed && timedOut))
23                 && (wc > 1 || workQueue.isEmpty())) {
24                 if (compareAndDecrementWorkerCount(c))
25                     return null;
26                 continue;
27             }
28 
29             try {
30                 Runnable r = timed ?    //跟据timed的值，选择使用阻塞方法获取队列中的值还是使用非阻塞方法获取
31                     workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
32                     workQueue.take();  //当核心线程调用这个方法时，会被阻塞，直到获取到任务，而不会返回null，然后被释放
33                 if (r != null)
34                     return r;
35                 timedOut = true;
36             } catch (InterruptedException retry) {
37                 timedOut = false;
38             }
39         }
40     }