你知道的越多,不知道的也越多!
想必大家对线程池都有了解吧,必须掌握的知识点有:构造函数里的入参列表、常见有哪几种线程池以及使用场景、为什么要用线程池、线程池接受到一个任务后的处理流程等等。
本文对以上问题不做说明了,主要是想讲明白其中的“线程复用”底层的原理。
啰嗦一句: 看源码不要一上来就死磕每一行代码(相信我死磕撑不了几分钟),先掌握大体思路,然后再仔细品。
相信以下这张图大家都很熟悉了:
其实就是线程池接受任务的处理规则。那图里的规则哪来的?废话,当然是大佬们看着源码画出来的啊!大佬不是我!!!
很好,其实“线程复用”和上述图里的任务队列有很大的关系。
线程复用不是简单的new Thread, 然后thread.start()这种思路。
而是将任务和线程解耦,任务交给队列存储、管理,然后线程不断地从队列里取任务,然后执行任务(Runnable实现)。执行的过程其实就是直接调用run方法,把run方法当作简单方法处理。
思路已经在上面暴露出来了,接下去只要跟下源码验证下啦~
咱直接进入到ThreadPoolExecutor#executor
方法,毕竟这个方法就是日常我们提交任务的方法。请注意抓重点!!!
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); }
我们先看下当线程数小于核心线程数的代码:
if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); }
这里workerCountOf
方法就是获取池子里的线程数的,到了后面你会知道一个Worker实例对应一个Thread,是不是顿时觉得人家这个方法名取得很有道理。
继续看重点,跟下addWorker
. 先说下两个入参:第一个入参就是业务代码写的每个任务,没啥说的;第二个参数:true-当前线程数是否小于核心线程数,小于则新增,大于则不新增;false-当前线程数是否小于最大线程数 。
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; }
建议这段代码拆成上下两部分,以“我是分割线”拆分:
Worker w = null; try { //新增Worker w = new Worker(firstTask); //Workder构造方法如下: Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); }
可以看出Worker实例化的时候,主要做了2件事情:
(1)把任务存在了firstTask变量里;
(2)内部变量thread实例化。细看下以下代码:
this.thread = getThreadFactory().newThread(this);
注意到this???那说明Worker肯定implement Runnable啦?
那,,,肯定要好好看下重写的run方法了。
接下去会发现直接调用了ThreadPoolExecutor#runWorker
既然这样了,那继续看下runWoker到底做了些什么?
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //获取提交的任务 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //重点:循环,要不取new Worker的任务,要不取getTask返回的任务 while (task != null || (task = getTask()) != null) { w.lock(); 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); } }
其实到这里: 我么可以看出Worker重新了run,其实就是一个循环,不断地取任务出来,然后执行任务。
那继续看getTask
方法,看从哪不断地取任务吧>>>>>>>
简直了,果然就是从队列里啊。那按理,提交的任务如果线程池满了,应该会进入到这个队列的哦。
其实到这里,我们还需要探索一个问题,既然Worker是实现了Runnable ,那什么时候,会调用Worker#run
真相其实上述代码已经贴出来啦,就是在addWorker -> new Worker 之后 。