快捷搜索:  汽车  科技

threadpoolexecutor 队列介绍(花15分钟浅谈ThreadPoolExecutor线程池及其源码)

threadpoolexecutor 队列介绍(花15分钟浅谈ThreadPoolExecutor线程池及其源码)public void execute(Runnable command) { // 首先判断任务是否为null,如果为null,抛出空指针异常 if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount a

什么是线程池

在我们的实际开发中,通常使用线程池创建线程资源。

而线程池本质上是一种池化技术,利用资源复用的思想,复用线程资源。

为什么使用线程池

那么,我们为什么要使用线程池呢?主要有以下两点原因:

  • 首先,使用线程池可以避免频繁创建和销毁线程所带来的性能开销。因为线程的创建会涉及到 CPU 上下文的切换以及内存的分配。
  • 其次,我们可以通过线程池参数控制线程的数量,避免无休止的创建线程带来的资源利用率过高的问题,从而起到资源保护的作用。
线程池的状态

threadpoolexecutor 队列介绍(花15分钟浅谈ThreadPoolExecutor线程池及其源码)(1)

在我们使用线程池时,一共涉及 5 种状态,其中包括:

  1. RUNNING :能接受新提交的任务,并且也能处理阻塞队列中的任务;
  2. SHUTDOWN:关闭状态,不再接受新提交的任务,但却可以继续处理阻塞队列中已保存的任务。在线程池处于 RUNNING 状态时,调用 shutdown() 方法会使线程池进入到该状态。(finalize() 方法在执行过程中也会调用 shutdown() 方法进入该状态);
  3. STOP:不能接受新任务,也不处理队列中的任务,会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时,调用 shutdownNow() 方法会使线程池进入到该状态;
  4. TIDYING:如果所有的任务都已终止了,workerCount (有效线程数) 为 0,线程池进入该状态后会调用 terminated() 方法进入 TERMINATED 状态。
  5. TERMINATED:在 terminated() 方法执行完后进入该状态,默认 terminated() 方法中什么也没有做。进入TERMINATED的条件如下:
  6. 线程池不是 RUNNING 状态;
  7. 线程池状态不是 TIDYING 状态或 TERMINATED 状态;
  8. 如果线程池状态是 SHUTDOWN 并且 workerQueue 为空;
  9. workerCount 为 0;
  10. 设置 TIDYING 状态成功。
线程池的分析及源码核心属性

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING 0)); // COUNT_BITS = 32 - 3 = 29 private static final int COUNT_BITS = Integer.SIZE - 3; // 工作线程的最大个数 // 00100000 00000000 00000000 00000000 - 1 // 000111111111111111111111111111111 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS; 复制代码

其中 ctl 属性维护着线程池的两个重要内容:

  1. 高 3 位维护着线程池的状态;
  2. 低 29 位维护着工作线程个数;
ctl 相关方法

// Packing and unpacking ctl // 获取线程池状态 private static int runStateOf(int c) { return c & ~CAPACITY; } // 获取工作线程数 private static int workerCountOf(int c) { return c & CAPACITY; } // 获取运行状态和工作线程数的值 private static int ctlOf(int rs int wc) { return rs | wc; } 复制代码ThreadPoolExecutor 构造方法

public ThreadPoolExecutor(int corePoolSize int maximumPoolSize long keepAliveTime TimeUnit unit BlockingQueue<Runnable> workQueue threadFactory threadFactory RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; } 复制代码

可以看出,在 ThreadPoolExecutor 的有参构造器中一共有 6 个参数:

  • corePoolSize:核心线程数
  • maximumPoolSize:最大线程数
  • keepAliveTime:线程池维护线程所允许的空闲时间,当线程池中的线程数量大于 corePoolSize 的时候,如果这时没有新的任务提交,核心线程外的线程不会立即销毁,而是会等待,直到等待的时间超过了 keepAliveTime
  • unit:时间单位
  • workQueue:等待队列,当任务提交时,如果线程池中的线程数量大于等于 corePoolSize 的时候,把该任务封装成一个 Worker 对象放入等待队列
  • threadFactory:它是 ThreadFactory 类型的变量,线程工厂,用来创建新的线程
  • handler:它是 RejectedExecutionHandler 类型的变量,表示线程池的饱和策略。如果阻塞队列满了并且没有空闲的线程,这时如果继续提交任务,就需要采取一种策略处理该任务。线程池提供了 4 种策略:
  • AbortPolicy:直接抛出异常,这是默认策略;
  • CallerRunsPolicy:用调用者所在的线程来执行任务;
  • DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
  • DiscardPolicy:直接丢弃任务;
核心思想

threadpoolexecutor 队列介绍(花15分钟浅谈ThreadPoolExecutor线程池及其源码)(2)

当有新任务提交的时候,首先会执行以下判断:

  1. 如果运行的线程少于 corePoolSize,则创建新线程来处理任务,即使线程池中的其他线程是空闲的;
  2. 如果线程池中的线程数量大于等于 corePoolSize 且小于 maximumPoolSize,则只有当 workQueue 满时才创建新的线程去处理任务;
  3. 如果设置的 corePoolSize 和 maximumPoolSize 相同,则创建的线程池的大小是固定的,这时如果有新任务提交,若 workQueue 未满,则将请求放入 workQueue 中,等待有空闲的线程去从 workQueue 中取任务并处理;
  4. 如果运行的线程数量大于等于 maximumPoolSize,这时如果 workQueue 已经满了,则通过 handler 所指定的策略来处理任务;

所以,任务提交时,判断的顺序为 corePoolSize –> workQueue –> maximumPoolSize。

源码execute 方法

public void execute(Runnable command) { // 首先判断任务是否为null,如果为null,抛出空指针异常 if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount and so prevents false alarms that would add * threads when it shouldn't by returning false. * * 2. If a task can be successfully queued then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped or start a new thread if there are none. * * 3. If we cannot queue task then we try to add a new * thread. if it fails we know we are shut down or saturated * and so reject the task. */ // 获取ctl int c = ctl.get(); // workerCountOf方法是获取工作线程数 // 如果当前工作线程数小于corePoolSize核心线程数 if (workerCountOf(c) < corePoolSize) { // 添加核心线程,其中true是根据corePoolSize判断,false是根据maximumPoolSize判断 if (addWorker(command true)) // 如果添加核心线程成功,直接return return; // 如果失败,再次获取ctl c = ctl.get(); } // 如果当前线程池状态是Running,并且任务添加到workQueue队列中成功 if (isRunning(c) && workQueue.offer(command)) { // 再次获取ctl int recheck = ctl.get(); // 再次判断线程池状态,如果线程池状态不是Running了,需要将刚才添加到workQueue队列中的任务移除 if (! isRunning(recheck) && remove(command)) // 移除workQueue队列中的任务 reject(command); // 如果工作线程数为0,则执行addWorker方法 // 第一个参数为null,表示在线程池中创建一个线程,但不去启动 // 第二个参数为false,代表是根据maximumPoolSize进行判断 // 如果判断workerCount大于0,则直接返回,在workQueue中新增的command会在将来的某个时刻被执行。 else if (workerCountOf(recheck) == 0) addWorker(null false); } // 如果执行到这里,有两种情况: // 1. 线程池已经不是RUNNING状态; // 2. 线程池是RUNNING状态,但workerCount >= corePoolSize并且workQueue已满。 // 这时,再次调用addWorker方法,但第二个参数传入为false,将线程池的有限线程数量的上限设置为maximumPoolSize; // 如果失败则拒绝该任务 else if (!addWorker(command false)) reject(command); } 复制代码

我们在执行 execute() 方法时,如果线程池状态一直是 RUNNING 的话:

  1. 如果 workerCount < corePoolSize,则创建并启动一个线程来执行新提交的任务;
  2. 如果 workerCount >= corePoolSize,且线程池内的阻塞队列未满,则将任务添加到该阻塞队列中;
  3. 如果 workerCount >= corePoolSize && workerCount < maximumPoolSize,且线程池内的阻塞队列已满,则创建并启动一个线程来执行新提交的任务;
  4. 如果 workerCount >= maximumPoolSize,并且线程池内的阻塞队列已满 则根据拒绝策略来处理该任务 默认的处理方式是直接抛异常。

这里要注意一下 addWorker(null false);,也就是创建一个线程,但并没有传入任务,因为任务已经被添加到workQueue中了,所以 worker 在执行的时候,会直接从 workQueue 中获取任务。所以,在 workerCountOf(recheck) == 0 时执行 addWorker(null false); 也是为了保证线程池在 RUNNING 状态下必须要有一个线程来执行任务。

addWorker 方法

addWorker() 方法的主要工作是在线程池中创建一个新的线程并执行。其中 firstTask 参数为新创建线程第一个需要执行的任务。core 参数如果为 true,则使用 corePoolSize 进行判断,false 则使用 maximumPoolSize。

private boolean addWorker(Runnable firstTask boolean core) { retry: for (;;) { // 获取ctl int c = ctl.get(); // 获取线程池状态 int rs = runStateOf(c); // Check if queue empty only if necessary. // 首先,如果rs >= SHUTDOWN,则表示线程池不再接收新任务 // 之后,再判断后面三个条件,只要有一个不满足,就返回false // 如果rs == SHUTDOWN,因为不再接收任务了,firstTask不是null,返回false // 如果firstTask是null,并且workQueue为空,也返回false // 因为workQueue已经为空了,不需要再创建线程了 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // 获取工作线程 int wc = workerCountOf(c); // 首先判断wc >= CAPACITY,如果大于等于工作线程最大值,返回false // 之后,如果core是true,则与corePoolSize比较,是false,则与maximumPoolSize比较 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 对工作线程执行 1操作, if (compareAndIncrementWorkerCount(c)) // 如果 1成功,跳出外层循环 // CAS方式,并发操作,只能有一个线程 1成功 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就是工作线程 Worker w = null; try { // 创建新的工作线程,将firstTask传入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()); // 如果线程池是RUNNING状态,或者线程池是SHUTDOWN状态且firstTask是null // 向线程池中添加线程 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // workers是HashSet,保存线程池中的所有工作线程 workers.add(w); int s = workers.size(); // 记录线程池中出现的最大线程数 if (s > largestPoolSize) largestPoolSize = s; // 将线程添加状态置为true,代表线程添加成功 workerAdded = true; } } finally { // 释放锁 mainLock.unlock(); } // 如果线程添加成功,启动线程,将线程启动状态置为true if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } 复制代码Worker 类

线程池中,每一个线程被封装成一个 Worker 对象,ThreadPool 维护的其实就是一组 Worker 对象。

我们可以看到 Worker 类继承了 AQS,实现了 Runnable 接口。其中包括两个属性,firstTask 和 thread。其中 firstTask 用来保存传入的任务,而 thread 是在调用构造方法时通过 ThreadFactory 类来创建的线程,用来处理任务。

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; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ 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); } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. protected boolean isHeldExclusively() { return getState() != 0; } protected boolean tryAcquire(int unused) { if (compareAndSetState(0 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } public boolean isLocked() { return isHeldExclusively(); } void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } } 复制代码runWorker 方法

因为 Worker 类实现了 Runnable 接口,因此,在 Worker 对象启动的时候,会调用重新的 run 方法,run 方法中调用了 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 { // 如果task为null,则通过getTask来获取任务 // 也就是Worker自身有携带任务,直接执行 // 如果Worker自身没有携带任务,那么就通过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 interrupt // 判断线程池状态是否大于等于STOP,如果是,要中断当前线程 if ((runStateAtLeast(ctl.get() STOP) || // DCL (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); } } 复制代码

runWorker 方法的执行过程:

  1. while 循环不断地通过 getTask() 方法获取任务;
  2. getTask() 方法从阻塞队列中取任务;
  3. 如果线程池正在停止,那么要保证当前线程是中断状态,否则要保证当前线程不是中断状态;
  4. 调用 task.run() 方法执行任务;
  5. 如果 task 为 null 则跳出循环,执行 processWorkerExit() 方法;
  6. runWork() 方法执行完毕,也代表着 Worker 中的 run 方法执行完毕,摧毁线程;
getTask 方法

getTask 方法主要用来在阻塞队列中获取任务。

private Runnable getTask() { // timeOut变量的值表示上次从阻塞队列中取任务时是否超时 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. // 如果线程池状态rs >= SHUTDOWN,也就是说是非RUNNING状态,那么再判断 // rs >= STOP,线程池是否正在stop // 阻塞队列为空 // 如果上面两个条件有一个满足,需要工作线程数-1,并返回null 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())) { // 工作线程数-1 if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 如果是非核心,走poll,拉取工作队列任务 // 如果是核心线程,走take一直阻塞,拉取工作队列任务 Runnable r = timed ? workQueue.poll(keepAliveTime TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } 复制代码processWorkerExit 方法

private void processWorkerExit(Worker w boolean completedAbruptly) { // 如果completedAbruptly为true,说明线程在执行时出现了异常,需要对工作线程数减1 // 执行正常的情况下,是在getTask方法中进行减1操作 if (completedAbruptly) // If abrupt then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 将当前工作线程完成的任务个数赋值给整个线程池中的任务数 completedTaskCount = w.completedTasks; // 从workers中移除,表示从线程池中移除了一个工作线程 workers.remove(w); } finally { mainLock.unlock(); } // 判断线程池是否可以中止,线程池状态是否发生变化 tryTerminate(); int c = ctl.get(); // 如果当前线程池状态小于STOP if (runStateLessThan(c STOP)) { // 判断线程池中的工作队列是否还有任务,并且工作线程是否还在 if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; if (workerCountOf(c) >= min) return; // replacement not needed } // 添加非核心空任务的线程处理工作队列中的任务 addWorker(null false); } }

链接:https://juejin.cn/post/7151428913325932580

猜您喜欢: