浅谈java线程池实现
再进入主题之前，我们先了解几个概念，对读源码有所帮助，对于线程池的运行状态，有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 }
所以，跟据上面的源码，核心线程不被释放的原因也被我们解决了，现在相信你对线程池的了解已经比较深刻了吧
这是博主的第一篇博客，如果写的有问题，希望大家在留言区进行指正，如果转载，请注明出处，尊重作者的辛苦付出，谢谢！