【Java线程池04】ThreadPoolExecutor的addWorker方法 |
您所在的位置：网站首页 › 线程池的execute方法 › 【Java线程池04】ThreadPoolExecutor的addWorker方法 |
【Java线程池04】ThreadPoolExecutor的addWorker方法
|
版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/u011637069/article/details/79593114 ThreadPoolExecutor的addWorker方法 前言【Java线程池01】Java线程池简介 此文中对Java线程池简介，包含线程池相关的几个类的关系等。 【Java线程池02】ThreadPoolExecutor类概述 此文介绍了ThreadPoolExecutor的构造函数、核心字段以及核心方法 【Java线程池03】ThreadPoolExecutor的execute方法执行流程 此文介绍了ThreadPoolExecutor的核心方法execute的逻辑。 addWorker方法源码比较长，看起来比较唬人，其实就做了两件事。1）才用循环CAS操作来将线程数加1；2）新建一个线程并启用。 源码如下： private boolean addWorker(Runnable firstTask, boolean core) { //（1）循环CAS操作，将线程池中的线程数+1. 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); //core true代表是往核心线程池中增加线程 false代表往最大线程池中增加线程 //线程数超标，不能再添加了，直接返回 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //CAS修改clt的值+1，在线程池中为将要添加的线程流出空间，成功退出cas循环，失败继续 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl //如果线程池的状态发生了变化回到retry外层循环 if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } //(2)新建线程，并加入到线程池workers中。 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //对workers操作要通过加锁来实现 final ReentrantLock mainLock = this.mainLock; w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { //细化锁的力度，防止临界区过大，浪费时间 mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int c = ctl.get(); int rs = runStateOf(c); //判断线程池的状态 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(); //修正largestPoolSize的值 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } //线程添加线程池成功，则开启新创建的线程 if (workerAdded) { t.start();//(3) workerStarted = true; } } } finally { //线程添加线程池失败或者线程start失败，则需要调用addWorkerFailed函数， //如果添加成功则需要移除，并回复clt的值 if (! workerStarted) addWorkerFailed(w); } return workerStarted; }逻辑分析 继承自AQS，具有锁的功能，实现了Runable接口，具有线程的功能。 private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; //线程池中正真运行的线程。通过我们指定的线程工厂创建而来 final Thread thread; //线程包装的任务。thread 在run时主要调用了该任务的run方法 Runnable firstTask; //记录当前线程完成的任务数 volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; //利用我们指定的线程工厂创建一个线程 this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); }由源码可知，Worker类的run方法实际上调用的还是ThreadPoolExecutor的runworker方法。下面将看一下ThreadPoolExecutor的runworker源代码和注释解析。 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(); //线程池处于stop状态或者当前线程被中断时，线程池状态是stop状态。 //但是当前线程没有中断，则发出中断请求 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) { wt.interrupt(); } try { //开始执行任务前的Hook，类似回调函数 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 { //任务执行后的Hook，类似回调函数 afterExecute(task, thrown); } } finally { //执行完毕后task重置，completedTasks计数器++，解锁 task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { //线程空闲达到我们设定的值时，Worker退出销毁。 processWorkerExit(w, completedAbruptly); } }大概意思就是当前任务不为null或者从队列中取的任务不为null时，worker线程就一直去执行任务。当无要执行的任务时，尝试回收线程。 runWorker函数中最重要的是getTask()，他不断的从阻塞队列中取任务交给线程执行。下面分析一下： private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); //如果线程池处于shutdown状态， //并且队列为空，或者线程池处于stop或者terminate状态， //在线程池数量-1，返回null，回收线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } //标识当前线程在空闲时，是否应该超时回收 boolean timed; for (;;) { int wc = workerCountOf(c); //如果allowCoreThreadTimeOut 为ture //或者当前线程数量大于核心线程池数目， //则需要超时回收 timed = allowCoreThreadTimeOut || wc > corePoolSize; //（1） //如果线程数目小于最大线程数目， //且不允许超时回收或者未超时， //则跳出循环，继续去阻塞队列中取任务（2） if (wc |
【本文地址】
今日新闻 |
推荐新闻 |