Java 线程池的复用原理

线程池的运行图

在线程池中，线程会从 workQueue 中读取任务来执行，最小的执行单位就是 Worker，Worker 实现了 Runnable 接口，重写了 run 方法，这个 run 方法是让每个线程去执行一个循环，在这个循环代码中，去判断workQueue中是否有任务，若有则直接去执行这个任务，因此线程数不会增加。

而在添加的线程的核心方法就是execute 中的addWorker方法：
首先看execute方法：

public void execute(Runnable command) {// 若任务为空，则抛 NPE，不能执行空任务if (command == null)throw new NullPointerException();int c = ctl.get();// 若工作线程数小于核心线程数，则创建新的线程，并把当前任务 command 作为这个线程的第一个任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}/*** 至此，有以下两种情况：* 1.当前工作线程数大于等于核心线程数* 2.新建线程失败* 此时会尝试将任务添加到阻塞队列 workQueue*/// 若线程池处于 RUNNING 状态，将任务添加到阻塞队列 workQueue 中if (isRunning(c) && workQueue.offer(command)) {// 再次检查线程池标记int recheck = ctl.get();// 如果线程池已不处于 RUNNING 状态，那么移除已入队的任务，并且执行拒绝策略if (! isRunning(recheck) && remove(command))// 任务添加到阻塞队列失败，执行拒绝策略reject(command);// 如果线程池还是 RUNNING 的，并且线程数为 0，那么开启新的线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}/*** 至此，有以下两种情况：* 1.线程池处于非运行状态，线程池不再接受新的线程* 2.线程处于运行状态，但是阻塞队列已满，无法加入到阻塞队列* 此时会尝试以最大线程数为界创建新的工作线程*/else if (!addWorker(command, false))// 任务进入线程池失败，执行拒绝策略reject(command);
}

第一步：如果工作线程小于核心线程，就会创建一个Worker线程，并把你提交的线程任务当做这个线程的第一个任务。

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

第二步：走到这表示工作线程已经超过核心线程了。那么就会把任务添加到workQueue

if (isRunning(c) && workQueue.offer(command)) {// 再次检查线程池标记int recheck = ctl.get();// 如果线程池已不处于 RUNNING 状态，那么移除已入队的任务，并且执行拒绝策略if (! isRunning(recheck) && remove(command))// 任务添加到阻塞队列失败，执行拒绝策略reject(command);// 如果线程池还是 RUNNING 的，并且线程数为 0，那么开启新的线程else if (workerCountOf(recheck) == 0)addWorker(null, false);
}

第三步：如果任务队列也满了，则会尝试创建非核心线程如果没有达到maximumPoolSize的话.如果超过maximumPoolSize就会执行拒绝策略

else if (!addWorker(command, false))// 任务进入线程池失败，执行拒绝策略reject(command);

再看addWorker方法：

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 ctlif (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 {// 把firstTask包装成Worker 线程池都是启动Worker线程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 startablethrow new IllegalThreadStateException();//把工作线程放到工作线程集合中去workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}// 如果成功添加了 Worker，就可以启动 Worker 了if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}

第一步：为判断这个线程是否能添加。

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 ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}
}

第二步：创建一个Worker线程，并且把当前任务传入当做这个线程的第一个任务

w = new Worker(firstTask);
final Thread t = w.thread;

第三步：把线程放到workers集合中去

if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();//把工作线程放到工作线程集合中去workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;
}

第四步：启动这个线程

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

再来看Worker构造函数：这里就是线程池创建线程的地方，把Worker作为线程创建

Worker(Runnable firstTask) {// 每个任务的锁状态初始化为-1，这样工作线程在运行之前禁止中断setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;// 把 Worker 作为 thread 运行的任务this.thread = getThreadFactory().newThread(this);
}

再看Worker的run方法：线程池启动的是Worker类，那么运行的run方法也是Worker 类的。run方法中又调用了runWorker方法
runWorker 有个循环一直在调用getTask方法。getTask方法就是获取workQueue任务队列的任务。获取到后就执行这个任务的run方法。

public void run() {// 线程池运行worker线程runWorker(this);
}final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 循环调用getTask 获取任务执行//getTask 的主要作用是阻塞从队列中拿任务出来，如果队列中有任务，那么就可以拿出来执行，如果队列中没有任务，这个线程会一直阻塞到有任务为止（或者超时阻塞）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 interruptif ((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);}
}private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}

总结：
1、线程池添加Runable其实只是调用run方法并没有真的启动这个Runable 线程
2、线程池中的线程是Worker类
3、Worker一直在自己run方法中去获取任务队列中的任务并执行他们的run方法。

Java 线程池的复用原理相关推荐

Java 线程池的工作原理
文章目录概念线程中的基本方法线程复用线程池的核心组件和核心类线程池的工作原理线程池中的workQueue任务队列直接提交队列(SynchronousQueue) 有界任务队列(Array ...
深入源码分析Java线程池的实现原理
转载自深入源码分析Java线程池的实现原理程序的运行,其本质上,是对系统资源(CPU.内存.磁盘.网络等等)的使用.如何高效的使用这些资源是我们编程优化演进的一个方向.今天说的线程池就是一种对 ...
Java线程池的实现原理，你清楚么？
点击关注公众号,实用技术文章及时了解来源:blog.csdn.net/u013332124/article/details/79587436 原理概述其实java线程池的实现原理很简单,说白了就是 ...
Java线程池使用与原理
线程池是什么? 我们可以利用java很容易创建一个新线程,同时操作系统创建一个线程也是一笔不小的开销.所以基于线程的复用,就提出了线程池的概念,我们使用线程池创建出若干个线程,执行完一个任务后,该线程 ...
全面解读Java线程池的工作原理
目录一.为什么引入线程池技术? 二.Executor框架 2.1 Runnable.Callable与Future接口 2.2 Executor接口 2.2.1 Executor 2.2.2 Exe ...
彻底搞懂Java线程池的工作原理
一.线程池的基础知识创建线程需要占用一定的操作系统资源,在高并发情况下,频繁的创建和销毁线程会大量消耗CPU和内存资源,对程序性能造成很大的影响.为了避免这一问题,Java提供了线程池(通过线程复用 ...
好文推荐：深入分析Java线程池的实现原理
线程是稀缺资源,如果被无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,合理的使用线程池对线程进行统一分配.调优和监控,有以下好处: 1.降低资源消耗: 2.提高响应速度: 3.提高线程的可管理 ...
深入源码，深度解析Java 线程池的实现原理
java 系统的运行归根到底是程序的运行,程序的运行归根到底是代码的执行,代码的执行归根到底是虚拟机的执行,虚拟机的执行其实就是操作系统的线程在执行,并且会占用一定的系统资源,如CPU.内存.磁盘.网 ...
给女朋友讲 : Java线程池的内部原理
文章持续更新,微信搜索「万猫学社」第一时间阅读. 关注后回复「电子书」,免费获取12本Java必读技术书籍. 餐厅的约会餐盘在灯光的照耀下格外晶莹洁白,女朋友拿起红酒杯轻轻地抿了一小口,对我 ...

Java 线程池的复用原理

线程池的运行图

Java 线程池的复用原理相关推荐

最新文章

热门文章