线程池的使用及ThreadPoolExecutor源码分析
一、线程
调度CPU资源的最小单位，线程模型分为 KLT 模型与 ULT 模型，JVM 使用的 KLT 模 型，Java 线程与 OS 线程保持1:1 的映射关系，也就是说有一个 Java 线程也会在操作系统里有一个对应的线程。详见前面的文章。
Java线程的生命状态 ：
- NEW 新建
- RUNNABLE 运行
- BLOCKED 阻塞
- WAITING 等待
- TIMED_WAITING 超时等待
- TERMINATED 终结
二、协程
协程 (纤程，用户级线程) ，目的是为了追求最大力度的发挥硬件性能和提升软件的速度，协程基本原理是：在某个点挂起当前的任务，并且保存栈信息，去执行另一个任务；等完成或达到某个条件时，再还原原来的栈信息并继续执行(整个过程线程不需要上下文切换)。
Java 原生不支持协程，在纯 java 代码里需要使用协程的话需要引入第三方包,如：quasar
三、线程池
线程是珍稀资源，如果被无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性， Java 中提供了线程池对线程进行统一分配、调优和监控。
如果并发的请求数量非常多，但每个线程执行的时间很短，这样就会频繁的创建和销毁线程，就会大大降低系统的效率。可能出现服务器在为每个请求创建新线程和销毁线程上花费的时间和消耗的系统资源要比处理实际的用户请求的时间和资源更多。 (例如一个任务只需要执行 10ms，可能创建销毁线程的时间比这更长)
项目中可能常见的Executors.newCachedThreadPool()等在阿里规范中是不被允许的(线程资源相当宝贵，违背了初衷，其使用 LinkedBlockingQueue 创建的是 Integer.MAX_VALUE 大小的队列，会堆积大量的请求，从而造成OOM)，建议在使用时自己 new 一个线程池。且一个线程池只执行一种职责，多个需求创建多个线程池。
线程池使用场景
- 单个任务处理时间比较短
- 需要处理的任务数量很大
为什么要用线程池
- 重用存在的线程，减少线程创建销毁的开销，提高性能
- 提高响应速度。当任务到达时，不需要等待线程创建就能就能立即执行
- 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控
线程池的使用
// 创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 5000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5)); // 模拟提交任务 for (int i=0;i<6;i++){ System.out.println(i); threadPoolExecutor.submit(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName()+"执行了任务"+); } },i); } // 结束线程池任务 threadPoolExecutor.shutdown(); // running => shutdown threadPoolExecutor.shutdownNow(); // running=> stop
Executor框架
Executor接口是线程池框架中最基础的部分，里面仅定义了一个用于执 Runnable 的 execute() 方法提供给子类重写。
ExecutorService 是 Excutor 的一个重要的子接口。里面定义了线程的具体行为：
| 方法 | 说明 |
|---|---|
| execute(Runnable command) | 执行Ruannable类型的任务 |
| submit(Runnable task) | 用来提交Callable或Runnable任务，并返回代表此任务的 Future 对象 |
| shutdown() | 完成已提交的任务后封闭线程池，不再接受新任务 |
| shutdownNow() | 停止所有正在履行的任务并封闭线程池 |
| isTerminated() | 测试是否所有任务都执行完毕了 |
| isShutdown() | 测试是否该ExecutorService已被关闭 |
线程池重要属性
/** * ctl是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段它包含两部分的信息: * 1.线程池状态 (runState) * 2.线程池内有效线程的数量 (workerCount)， * 使用了Integer类型来保存，高3位保存runState，低29位保存 workerCount * 具体使用见下面源码 */ private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; // 使用高三位 private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 使用低29位 // 线程池存在的五种状态 // runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS; // 高三位111 private static final int SHUTDOWN = 0 << COUNT_BITS; // 高三位000 private static final int STOP = 1 << COUNT_BITS; // 高三位001 private static final int TIDYING = 2 << COUNT_BITS; // 高三位010 private static final int TERMINATED = 3 << COUNT_BITS; // 高三位011 // Packing and unpacking ctl 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; }
线程池的五种状态说明
-
RUNNING
- 状态说明： 能够接收新任务，以及对已添加的任务进行处理
- 状态切换： 线程池的初始化状态，一旦被创建就处于该状态，且线程池中任务数量为0
-
SHUTDOWN
- 状态说明： 不接收新任务，但能处理已添加的任务
- 状态切换： 调用线程池的 shutdown() 接口时，线程池由RUNNING => SHUTDOWN
-
STOP
- 状态说明： 不接收新任务，不处理已添加的任务，并且会中断正在处理的任务
- 状态切换： 调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) => STOP
-
TIDYING
- 状态说明： 当所有的任务已终止，ctl 记录的”任务数量“为0，线程池会变为 TIDYING 状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated() 在ThreadPoolExecutor类中是空的，若用户想在线程池变为 TIDYING 时，进行相应的处理，可以通过重载 terminated() 函数来实现
- 状态切换： 当线程池在 SHUTDOWN 状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN => TIDYING。 当线程池在 STOP 状态下，线程池中执行的任务为空时，就会由STOP => TIDYING
-
TERMINATED
- 状态说明： 线程池彻底已终止
- 状态切换： 线程池处在 TIDYING 状态时，执行完 terminated() 之后，就会由 TIDYING => TERMINAT
线程池的创建及其七大参数
线程池的创建及任务提交
// 创建 new 构造函数 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } // 任务提交 // 1.提交任务无返回值 public void execute(){ ... } // 2\. 任务执行完成后有返回值 public Future<?> submit(){ ... }
核心参数说明
int corePoolSize 核心线程数
- 当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize。如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行。如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程
int maximumPoolSize 最大线程数
- 如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize
long keepAliveTime 线程最大空闲时间
- 线程池维护线程所允许的空闲时间。当线程池中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了keepAliveTime
Timeunit unit 最大空闲等待的时间单位
- keepAliveTime 的时间单位
BlockingQueue<Runnable> workQueue 等待(阻塞)队列（存放未来得及执行的任务）
- 用来保存等待被执行的任务的阻塞队列，且任务必须实现 Runable 接口。
- ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务
- LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene
- SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于 LinkedBlockingQuene
- PriorityBlockingQuene：具有优先级的无界阻塞队列
ThreadFactory threadFactory 创建线程的工厂
- 用来创建新线程。默认使用 Executors.defaultThreadFactory() 来创建线程。使用默认的ThreadFactory来创建线程时，会使新创建的线程具有相同的 NORM_PRIORITY 优先级并且是非守护线程，同时也设置了线程的名称
RejectedExecutionHandler handler 拒绝策略
- 线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务
拒绝策略
4 种拒绝默认的策略：
- AbortPolicy 默认策略，直接抛异常
- CallerRunsPolicy 由当前提交任务的线程执行任务（调用者线程），而不给线程池中线程执行
- DiscardOldestPolicy 丢弃掉旧的任务（队头任务），并执行当前任务
- DiscardPolicy 直接丢弃任务
自定义拒绝策略：
不希望任务被丢弃可以使用无界队列，如果还要考虑JVM容量，可以重写拒绝策略。比如将丢到中间件中(Redis等)，比如监控线程池中队列少于50%时，从Redis中读之前容纳不下的任务。重写拒绝策略只需要实现 RejectedExecutionHandler 接口并重写 rejectedExecution() 方法即可。
线程数量设置（仅做参考）
CPU 密集型： CPU核数 + 1 不会怎么停顿
I/O密集型： 两倍 CPU核数 + 1 会停顿，读文件等
最佳线程数: CPU 核数 * [1 + (I/O耗时 / CPU耗时)]
线程池监控方法
public long getTaskCount() // 获取线程池已执行与未执行的任务总数 public long getCompletedTaskCount()// 获取已完成的任务数 public int getPoolSize() // 线程池当前的线程数 public int getActiveCount() // 线程池中正在执行任务的线程数量 线程池原理
思考：线程池中阻塞队列的作用？
- 一般队列只能保证一个有限长度的缓冲区，如果超过缓冲长度，就无法保留当前任务，通过阻塞队列可以通过阻塞保留当前想要继续入队的任务
- 阻塞队列自带阻塞和唤醒功能，不需要额外处理，无任务执行时线程池利用阻塞队列 take 方法挂起，从而维持核心线程存活，不至于一直占用 CPU 资源
思考：为什么先添加阻塞队列而不是先创建最大线程？
在创建新线程的时候，需要获取全局锁，这时就会阻塞其他线程，影响整体效率。最大线程如果后续销毁也会影响效率。
源码分析
线程工作线程生命周期
excute方法
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); // clt 记录着runState 和 workerCount int c = ctl.get(); /* * workerCountOf 方法取出低29位的值，即获取当前活动的线程数； * 如果当前活动线程数小于 corePoolSize，则新建一个线程放入线程池中 * 并把任务添加到该线程中 */ if (workerCountOf(c) < corePoolSize) { /* * addWorker 中的第二个参数表示限制添加线程的数量是根据corePoolSize来判断 * 还是maximumPoolSize来判断： * 1.如果为true，根据corePoolSize来判断 * 2.如果为false，则根据maximumPoolSize来判断 */ if (addWorker(command, true)) return; // 如果添加失败，则重新获取ctl值 c = ctl.get(); } // 如果当前线程池是运行状态 并且 任务添加到队列成功 if (isRunning(c) && workQueue.offer(command)) { // 重新获取ctl值 int recheck = ctl.get(); /* * 再次判断线程池的运行状态，如果不是运行状态，由于之前已经把command添加 * 到workQueue中了，这时需要移除该command执行过后通过handler使用拒绝策 * 略对该任务进行处理，整个方法返回 */ if (! isRunning(recheck) && remove(command)) reject(command); /* * 获取线程池中的有效线程数，如果数量是0，则执行addWorker方法 * 这里传入的参数表示： * 1\. 第一个参数为null，表示在线程池中创建一个线程，但不去启动，在任务 * 队列中取任务执行 * 2\. 第二个参数为false，将线程池的线程数量的上限设置为maximumPoolSize， * 添加线程时根据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); }
用一张图来表示就是如下流程：
源码解读： 首先当一个任务来后，先检查核心线程数是否满，未满则创建核心线程，此任务完成后，即使再来新的任务也不会再使用刚刚创建的核心线程，而是继续创建核心线程调用后续任务。然后当核心线程数满后，不管核心线程是否都被占用都会将任务放置到等待队列，以后线程池执行任务都从等待队列中去取，核心线程都被占用后会检查是否达到最大线程数，未达到创建最大线程，最大线程数满后会执行拒绝策略。值得注意的是线程池并未标识核心非核心线程，仅依靠线程的数量控制。
addWorker方法
用于在线程池中创建一个新的线程并执行，它有两个参数，firstTask 参数用于指定新增的线程执行的第一个任务，core 参数为 true 表示在新增线程时会判断当前活动线程数是否少于 corePoolSize，false 表示新增线程前需要判断当前活动线程数是否少于 maximumPoolSize.
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); // 获取运行状态 int rs = runStateOf(c); /* * 如果rs >= SHUTDOWN，则表示此时不再接收新任务； * 接着判断以下3个条件，只要有1个不满足，则返回false： * 1\. rs == SHUTDOWN，这时表示关闭状态，不再接受新提交的任务， * 但却可以继续处理阻塞队列中已保存的任务 * 2\. firsTask 为空 * 3\. 阻塞队列不为空 * * a.当 rs == SHUTDOWN 时不会接受新提交的任务，所以在firstTask不为空 * 的时候会返回false； * b.如果firstTask为空，并且workQueue也为空，则返回false， * 因为队列中已经没有任务了，不需要再添加线程了 */ // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // 获取线程数 int wc = workerCountOf(c); /* * 如果wc超过CAPACITY，也就是ctl的低29位的最大值（二进制是29个1）， * 返回false。这里的core是addWorker方法的第二个参数，如果为true表 * 示根据corePoolSize来比较，如果为false则根据maximumPoolSize来比较。 */ if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 尝试增加workerCount，如果成功，则跳出第一个for循环 if (compareAndIncrementWorkerCount(c)) break retry; // 如果增加workerCount失败，则重新获取ctl的值 c = ctl.get(); // Re-read ctl // 如果当前的运行状态不等于rs，说明状态已被改变，返回第一个for循环继续执行 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 { // 根据firstTask来创建Worker对象 w = new Worker(firstTask); // 每一个Worker对象都会创建一个线程 final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int rs = runStateOf(ctl.get()); /* * rs < SHUTDOWN表示是RUNNING状态； * 如果rs是RUNNING状态或者rs是SHUTDOWN状态并且firstTask为null， * 向线程池中添加线程。因为在SHUTDOWN时不会在添加新的任务，但还是 * 会执行workQueue中的任务 */ 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(); // largestPoolSize记录着线程池中出现过的最大线程数量 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 启动线程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
Worker类
ThreadPoolExecutor 中定义了一个内部类 Worker，线程池中的每一个线程被封装成一个Worker对象，ThreadPool 维护的其实就是一组 Worker 对象
Worker源码，部分逻辑已省略：
// Worker类继承了AQS，并实现了Runnable接口 private final class Worker extends AbstractQueuedSynchronizer implements Runnable { // 调用构造方法时通过ThreadFactory来创建的线程,是用来处理任务的线程 final Thread thread; /** Initial task to run. Possibly null. */ // 保存传入的任务 Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * 构造方法 * 传入任务，创建线程 * newThread方法传入的参数是this，因为Worker本身继承了Runnable接口，也就是一个线程， * 所以一个Worker对象在启动的时候会调用Worker类中的run方法调用RunWorker() */ Worker(Runnable firstTask) { /* * AQS中默认的state是0，如果刚创建了一个Worker对象，还没有执行任务时，这时就 * 不应该被中断，禁止在执行任务前对线程进行中断，在runWorker方法中也会先调用 * Worker对象的unlock方法将state设置为0 */ 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); } /** * AQS来实现独占锁的功能 */ protected boolean tryAcquire(int unused) { //cas修改state，不可重入 if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } }
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会被置空，跳出循环 // 如果task为空，则通过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); } }
getTask方法
作用：从阻塞队列中取任务
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. /* * 如果线程池状态rs >= SHUTDOWN，也就是非RUNNING状态，再进行以下判断： * 1\. rs >= STOP，线程池是否正在stop； * 2\. 阻塞队列是否为空。 * 如果以上条件满足，则将workerCount减1并返回null。 * 因为如果当前线程池状态的值是SHUTDOWN或以上时，不允许再向阻塞队列中添加任务。 */ if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? /* * 根据timed来判断，如果为true，则通过阻塞队列的poll方法进行超时控制； * 如果在keepAliveTime时间内没有获取到任务，则返回null； * 否则通过take方法，如果这时队列为空，则take方***阻塞直到队列不为空。 * */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /* * wc > maximumPoolSize的情况是因为可能在此方法执行阶段同时 * 执行setMaximumPoolSize方法； * timed && timedOut 如果为true，表示当前操作需要进行超时控制，并且上次 * 从阻塞队列中获取任务发生了超时,然后判断，如果有效线程数量大于1，或阻塞队 * 列为空，那么尝试将workerCount减1； * 如果减1失败，则返回重试。 * 如果wc == 1时，说明当前线程是线程池中唯一的一个线程 */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { Runnable r = timed ? // 阻塞队列，当队列为空，条件不满足，阻塞，等待多长时间还为空，返回null workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; // 如果 r == null，说明已经超时，timedOut设置为true timedOut = true; } catch (InterruptedException retry) { // 如果获取任务时当前线程发生了中断，则设置timedOut为false并返回循环重试 timedOut = false; } } }
processWorkerExit方法
private void processWorkerExit(Worker w, boolean completedAbruptly) { /* * 如果completedAbruptly值为true，说明线程执行时出现了异常，需要将workerCount减1； * 如果线程执行时没有出现异常，说明在getTask()方法中已经已经对workerCount进行了减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(); /* * 当线程池是RUNNING或SHUTDOWN状态时，如果worker是异常结束，那么会直接addWorker； * 如果allowCoreThreadTimeOut=true，并且等待队列有任务，至少保留一个worker； * 如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize。 */ 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); } }
processWorkerExit 执行完之后，工作线程被销毁，工作线程生命周期结束，从 execute 方法开始，Worker 使用 ThreadFactory 创建新的工作线程，runWorker 通过 getTask 获取任务，然后执行任务，如果 getTask 返回 null，进入 processWorkerExit 方法，整个线程结束。
我的总结
ThreadPoolExecutor 定义了AtomicInteger ctl ，高三位表示线程池状态，低29位保存有效线程的数量，通过这个变量去判断线程的状态及数量，它里面也没有额外的标志去判断是否核心线程，仅仅通过数量来控制。ThreadPoolExecutor 内部定义了一个内部类 Worker继承了AQS类，实现了Runnable接口，线程池中的每一个线程都会被封装成一个Worker对象，里面定义了Runnable firstTask 和 Thread thread;我们传入的任务就会被有参构造保存起来，并通过线程工厂new一个线程。当我们执行addWorker()添加一个线程后会调用thread.start()启动线程调用里面的run()方法，然后run()方法里面就做了一件事，就是调用runWorker()方法。它里面会不断的循环通过一个getTask()方法不断的调用poll()方法从队列中拿任务，直到任务全部被执行完成，task被置空跳出循环。最后执行一个processWorkerExit()方法销毁工作线程
四、定时任务线程池
ScheduledThreadPoolExecutor 用于指定的延迟后运行命令，或者定期执行命令。Timer 也可以实现相关需求，但 ScheduledThreadPoolExecutor 功能更强大、更灵活。在阿里规范中是禁止使用 Timer 的。
Timer timer = new Timer()
- 1.Timer 对应的是单个后台线程。而 ScheduledThreadPoolExecutor 可以在构造函数中指定多个对应的后台线程数。
- 2.Timer 无异常处理，且出现异常不会重新创建线程，因此一旦抛出非受检异常，该线程会立即终止。ScheduledThreadPoolExecutor 可以进行异常的处理，且出现异常时会重新创建新的线程执行任务。
- 3.Timer 执行程序是有可能延迟误差较 ScheduledThreadPoolExecutor 大很多。
- 4.使用定时任务线程池时必须捕获异常进行处理，ScheduledThreadPoolExecutor 底层在处理异常时会将任务丢弃，造成线程池中无任务可以运行。
Timer 的使用
//5000表示第一次执行任务延迟时间，100表示以后每隔多长时间执行一次run里面的任务，单位（毫秒） Timer timer = new Timer(); timer.schedule(new TimerTask() { @Override public void run() { log.e("time:"); } }, 5000, 100);
ScheduledThreadPoolExecutor 定时线程池类的类关系及结构图
存储结构
ScheduledThreadPoolExecutor 使用 无界队列 DelayQueue 来存储等待的任务，DelayQueue 内部封装了一个 PriorityQueue，会将任务根据 time 的先后时间排序，若 time 相同则根据 sequenceNumber 排序。内部采用最小堆结构存储数据（最小的元素永远在堆顶，只需要关心最小的堆顶元素，不需要关心其他元素位置及顺序，但要保证子节点的值要比父节点大。时间复杂度O(n) = lgn）
任务的提交
定时任务线程池内部提供了三种方法来提交任务：
- schedule：任务在指定延迟时间到达后触发，只会执行一次
- scheduledAtFixedRate：任务在指定延迟时间到达后触发，会执行多次，可能造成任务积压
- scheduledWithFixedDelay：前一个任务执行完成的情况下，任务在指定延迟时间到达后触发，会执行多次，不会造成任务积压
使用
// 创建一个只有一个线程的定时任务线程池 ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1); scheduledThreadPoolExecutor.schedule(() -> { log.info("延迟 5s 执行方法，且只执行一次"); return 1; }, 5000, TimeUnit.MILLISECONDS); // 发送心跳，service1->service2，每次过5s，发送一个心跳，证明s2可用 // 任务在指定延迟时间到达后触发 // 注意捕获异常 scheduledThreadPoolExecutor.scheduleWithFixedDelay(() -> { log.info("scheduleWithFixedDelay send heart beat"); long starttime = System.currentTimeMillis(), nowtime = starttime; while ((nowtime - starttime) < 5000) { nowtime = System.currentTimeMillis(); try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } log.info("scheduleWithFixedDelay task over...."); }, 1000, 2000, TimeUnit.MILLISECONDS); // 发送心跳，service1->service2，每次过5s，发送一个心跳，证明s2可用 // 前一个任务执行完成的情况下，任务在指定延迟时间到达后触发 // 注意捕获异常 scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> { log.info("scheduleWithFixedDelay send heart beat"); long starttime = System.currentTimeMillis(), nowtime = starttime; while ((nowtime - starttime) < 5000) { nowtime = System.currentTimeMillis(); try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } log.info("scheduleWithFixedDelay task over...."); }, 1000, 2000, TimeUnit.MILLISECONDS);
工作线程的执行过程：
- 工作线程会从 DelayQueue 队列取已经到期的任务去执行；
- 执行结束后重新设置任务的到期时间，再次放回 DelayQueue
DelayQueue 封装了一个 PriorityQueue，PriorityQueue 会对队列中的 ScheduledFutureTask 进行排序。
源码解析
/** * 提交任务 */ public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); if (delay <= 0) throw new IllegalArgumentException(); // 嵌套结构，首先把用户提交的任务包装成 ScheduledFutureTask ScheduledFutureTask<Void> sft = new ScheduledFutureTask<Void>(command, null, triggerTime(initialDelay, unit), unit.toNanos(-delay)); // 调用decorateTask进行包装，该方法是留给用户去扩展的，默认是个空方法 RunnableScheduledFuture<Void> t = decorateTask(command, sft); sft.outerTask = t; // 提交包装好的任务 delayedExecute(t); return t; } /** * 包装任务 */ ScheduledFutureTask(Runnable r, V result, long ns) { super(r, result); this.time = ns; this.period = 0; this.sequenceNumber = sequencer.getAndIncrement(); } public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; // ensure visibility of callable } /** * 提交包装好的任务 */ private void delayedExecute(RunnableScheduledFuture<?> task) { //如果线程池已经关闭，则使用拒绝策略把提交任务拒绝掉 if (isShutdown()) reject(task); else { //与ThreadPoolExecutor不同，这里直接把任务加入延迟队列 super.getQueue().add(task);//使用用的DelayedWorkQueue //如果当前状态无法执行任务，则取消 if (isShutdown() && !canRunInCurrentRunState(task.isPeriodic()) && remove(task)) task.cancel(false); else //这里是增加一个worker线程，避免提交的任务没有worker去执行 //原因就是该类没有像ThreadPoolExecutor一样，woker满了才放入队列 ensurePrestart(); } } /* * 排序 * 首先按照time排序，time小的排在前面，time大的排在后面，如果时间相同，执行先提交的任务 * sequenceNumber 越小，越先提交 */ public int compareTo(Delayed other) { if (other == this) // compare zero if same object return 0; if (other instanceof ScheduledFutureTask) { ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other; long diff = time - x.time; if (diff < 0) return -1; else if (diff > 0) return 1; else if (sequenceNumber < x.sequenceNumber) return -1; else return 1; } long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS); return (diff < 0) ? -1 : (diff > 0) ? 1 : 0; }
run 方法是调度 task 的核心，task的执行实际上是 run 方法的执行的：
public void run() { boolean periodic = isPeriodic(); //如果当前线程池已经不支持执行任务，则取消 if (!canRunInCurrentRunState(periodic)) cancel(false); //如果不需要周期性执行，则直接执行run方法然后结束 else if (!periodic) ScheduledFutureTask.super.run(); //如果需要周期执行，则在执行完任务以后，设置下一次执行时间 else if (ScheduledFutureTask.super.runAndReset()) { setNextRunTime(); // 计算下次执行该任务的时间 reExecutePeriodic(outerTask); //重复执行任务 } } /** * 周期性执行任务，与 delayedExecute 类似，捕获reject当前任务，且传入 * 的任务一定是周期性任务 */ void reExecutePeriodic(RunnableScheduledFuture<?> task) { if (canRunInCurrentRunState(true)) { super.getQueue().add(task); if (!canRunInCurrentRunState(true) && remove(task)) task.cancel(false); else ensurePrestart(); } }#Java##程序员##计算机##Java学习#
投递粉笔网等公司10个岗位 >
查看12道真题和解析