前言
Java线程池中提交任务运行,通常使用execute()
方法就足够了。那如果想要实现在主线程中阻塞获取线程池任务运行的结果,该怎么办呢?答案是用submit()
方法提交任务。这也是面试中经常被问到的一个知识点,execute()
和submit()
提交任务的的区别是什么?底层是如何实现的?
案例演示
现在我们通过简单的例子演示下submit()方法的妙处。
@Test public void testSubmit() throws ExecutionException, InterruptedException { // 创建一个核心线程数为5的线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue(50)); // 创建一个计算任务 Callable myTask = new Callable() { @Override public Integer call() throws Exception { int result = 0; for (int i = 0; i future = threadPoolExecutor.submit(myTask); Integer sum = future.get(); log.info("get submit result: [{}]", sum); // use sum do other things }
运行结果:
主线程的确阻塞等待线程返回。
Future类API
我们看到用submit提交任务最后返回一个Future对象,Future表示异步计算的结果。那它都提供了什么API呢?
方法 | 说明 |
---|---|
V get() | 等待任务执行完成,然后获取其结果。 |
V get(long timeout, TimeUnit unit) | 等待获取任务执行的结果,如果任务超过一定时间没有执行完毕,直接返回,抛出异常,不会一直等待下去。 |
boolean isDone() | 如果此任务已完成,则返回true。完成可能是由于正常终止、异常或取消——在所有这些情况下,该方法都将返回true。 |
boolean isCancelled() | 如果该任务在正常完成之前被取消,则返回true。 |
boolean cancel(boolean mayInterruptIfRunning) | 试图取消此任务的执行。1. 如果任务已经完成、已经取消或由于其他原因无法取消,则此尝试将失败。 |
- 如果在调用cancel时此任务尚未启动,则此任务不应运行。
- 如果任务已经开始,那么mayInterruptIfRunning参数确定是否应该中断执行此任务的线程以试图停止该任务。 |
和execute区别
从功能层面,我们已经很明白他们最大区别,
-
execute()
方式提交任务没有返回值,直接线程中池异步运行任务。 -
submit()
方式提交任务有返回值Future, 调用get方法可以阻塞调用线程,等待任务运行返回的结果。
那从源码层面,二者又有什么区别和联系呢?
我们看下submit()
提交的入口方法,代码如下:
// AbstractExecutorService#submit public Future submit(Callable task) { // 判空处理 if (task == null) throw new NullPointerException(); // 将提交的任务包装成RunnableFuture RunnableFuture ftask = newTaskFor(task); // 最终还是调用execute方法执行任务 execute(ftask); return ftask; }
殊途同归,最终都是调用execute()
方法,只不过submit()
方法在调用前做一层包装,将任务包装成RunnableFuture
对象。
关于线程池中execute()
方法提交的流程和原理实现不理解的,强烈建议先学习这篇文章:Java线程池源码深度解析。
原理实现
本节内容我们聚焦在submit()方法的实现原理。
我们先思考下,如果让我们设计实现调用get阻塞知道线程返回结果,要考虑哪些方面呢?
- 任务是否执行结束或者执行出错等情况,是不是需要有个状态位标记?
- 任务的执行结果如何保存?
- 如果任务没有执行结束,如何阻塞当前线程,
LockSupport.park()
是一种方式。 - 如果有多个外部线程获取get,是不是应该也要把外部线程存下来,怎么存?因为后面任务执行完后需要唤醒他们。
带着这些问题和基本思路我们看下jdk8中是如何实现的。
RunnableFuture类介绍
submit()
方法中调用newTaskFor()
方法获取RunnableFuture
对象。
// AbstractExecutorService#newTaskFor protected RunnableFuture newTaskFor(Callable callable) { // 调用FutureTask的构造方法返回RunnableFuture对象 return new FutureTask(callable); }
FutureTask
类结构图如下:
FutureTask
是一个异步计算任务,包装了我们外部提交的任务。
- 实现了Runnable接口
- 实现了Future接口,该接口封装了任务结果的获取、任务是否结束等接口。
RunnableFuture类重要属性
1.任务运行状态state
// 存储当前任务运行状态 private volatile int state; // 当前任务尚未执行 private static final int NEW = 0; // 当前任务正在结束,尚未完全结束,一种临界状态 private static final int COMPLETING = 1; // 当前任务正常结束 private static final int NORMAL = 2; // 当前任务执行过程中发生了异常 private static final int EXCEPTIONAL = 3; // 当前任务被取消 private static final int CANCELLED = 4; // 当前任务中断中 private static final int INTERRUPTING = 5; // 当前任务已中断 private static final int INTERRUPTED = 6;
可能的状态转换有如下几种:
- NEW -> COMPLETING -> NORMAL
- NEW -> COMPLETING -> EXCEPTIONAL
- NEW -> CANCELLED
- NEW -> INTERRUPTING -> INTERRUPTED
2.真正要执行的任务callble
// 存放真正提交的原始任务 private Callable callable;
3.存放执行结果outcome
返回的结果或从get()中抛出的异常 private Object outcome;
4.当前正在运行任务的线程runner
//当前任务被线程执行期间,保存当前任务的线程对象引用 private volatile Thread runner;
5.调用get获取任务结果的等待线程集合waiters
//因为会有很多线程去get当前任务的结果,所以这里使用了一种stack数据结构来保存 private volatile WaitNode waiters; static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread.currentThread(); } }
数据结构如下图:
RunnableFuture类构造方法
public FutureTask(Callable callable) { if (callable == null) throw new NullPointerException(); // 设置要执行的任务 this.callable = callable; // 初始化时任务状态为NEW this.state = NEW; }
任务执行run()原理
submit()
方法最终调用线程池的execute()
方法,而execute()
方法会创建出”工人”Worker
对象,调用runWorker()
方法,它主要是执行外部提交的任务,也就是这里的FutureTask
对象的run()
方法, 我们重点看下run()
方法。
FutureTask#run()
开始执行任务。
它主要的功能是完成包装的callable的call方法执行,并将执行结果保存到outcome中,同时捕获了call方法执行出现的异常,并保存异常信息,而不是直接抛出。
public void run() { // 状态机不为NEW表示执行完成或任务被取消了,直接返回 // 状态机为NEW,同时将runner设置为当前线程,保证同一时刻只有一个线程执行run方法,如果设置失败也直接返回 if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable c = callable; // 取出原始的任务检测不为空 且 再次检查状态为NEW(双重校验) if (c != null && state == NEW) { // 任务运行的结果 V result; // 任务是否运行是否正常, true:正常, false-异常 boolean ran; try { // 任务执行,将结果返回给result result = c.call(); // 设置任务运行正常 ran = true; } catch (Throwable ex) { // 任务运行报错的情况 // 设置结果为空 result = null; // 设置任务运行异常标记 ran = false; // 任务执行抛出异常时,保存异常信息,而不直接抛出 setException(ex); } // 执行成功则保存结果 if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() // 执行完成后设置runner为null runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts // 获取任务状态 int s = state; // 如果被置为了中断状态则进行中断的处理 if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
FutureTask#set()
方法处理正常执行的运行结果
setException()方法主要完成做下面的工作。
- 将执行结果保存到outcom变量中
- FutureTask的状态从NEW修改为NORMAL
- 唤醒阻塞在waiters队列中请求get的所有线程
protected void set(V v) { // 将状态由NEW更新为COMPLETING if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { // 保存任务的结果 outcome = v; // 更新状态的最终状态-NORMAL UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state // 通用的完成操作,主要作用就是唤醒阻塞在waiters队列中请求get的线程 finishCompletion(); } }
FutureTask#setException()
方法处理执行异常的结果
setException()方法主要完成做下面的工作。
- 将异常信息保存到outcom变量中
- FutureTask的状态从NEW修改为EXCEPTIONAL
- 唤醒阻塞在waiters队列中请求get的所有线程
// FutureTask#setException protected void setException(Throwable t) { // 将状态由NEW更新为COMPLETING if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { // 将异常信息保存到输出结果中 outcome = t; // 更新状态机处理异常的最终状态-EXCEPTIONAL UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state // 通用的完成操作,主要作用就是唤醒阻塞在waiters队列中请求get的线程 finishCompletion(); } }
这里的finishCompletion()
唤醒我们在后面讲解,上面的整个逻辑可以用一张图表示:
任务结果获取get()原理
其他线程可以调用get()
方法或者超时阻塞方法get(long timeout, TimeUnit unit)
获取任务运行的结果。
FutureTask#get()
方法是获取任务执行结果的入口方法。
// 阻塞获取任务结果 public V get() throws InterruptedException, ExecutionException { int s = state; // 任务还没有执行完成,通过awaitDone方法进行阻塞等待 if (s
FutureTask#awaitDone()
方法阻塞等待任务执行结束
该方法主要完成下面的工作:
- 判断任务是否运行结束,结束的话直接返回运行状态
- 如果任务没有结果,将请求线程阻塞
- 请求线程阻塞时,会创建一个waiter节点,然后加入到阻塞等待的栈中
// 线程阻塞等待方法, timed等于 true表示阻塞等待有时间限制nanos, false表示没有,一直阻塞 private int awaitDone(boolean timed, long nanos) throws InterruptedException { // 计算阻塞超时时间点 final long deadline = timed ? System.nanoTime() + nanos : 0L; WaitNode q = null; // 表示q是否添加到waiters栈中,默认false boolean queued = false; // 自旋操作 for (;;) { // 如果阻塞线程被中断则将当前线程从阻塞队列中移除 if (Thread.interrupted()) { // 从waiters栈中移除WaitNode, removeWaiter(q); // 返回中断移除 throw new InterruptedException(); } // 获取任务的状态 int s = state; // 如果任务的状态大于COMPLETING,表示线程运行结束了,直接返回 if (s > COMPLETING) { // 任务已经完成时直接返回结果 if (q != null) q.thread = null; // 返回状态 return s; } // 如果任务状态是COMPLETING else if (s == COMPLETING) // 如果任务执行完成,但还差最后一步最终完成,则让出CPU给任务执行线程继续执行 Thread.yield(); // 如果任务状态小于COMPLETING,说明任务还在运行中 // 如果q为空的情况 else if (q == null) // 新进来的线程添加等待节点 q = new WaitNode(); // 如果任务还在运行中并且当前线程节点还不在waiters栈中,那么就加入 else if (!queued) // 上一步节点创建完,还没将其添加到waiters栈中,因此在下一个循环就会执行此处进行入栈操作,并将当前线程的等待节点置于栈顶 queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); // 如果任务还在运行中并且timed为true,表示有超时限制 else if (timed) { // 如果设置了阻塞超时时间,则进行检查是否达到阻塞超时时间,达到了则删除当前线程的等待节点并退出循环返回,否则继续阻塞 nanos = deadline - System.nanoTime(); // 如果nanos小于等于0 if (nanos
FutureTask#report
方法解析返回任务结果
// 获取任务结果方法:正常执行则直接返回结果,否则抛出异常 private V report(int s) throws ExecutionException { Object x = outcome; // 如果状态是正常情况 if (s == NORMAL) // 直接返回 return (V)x; // 如果状态是取消了,抛出异常 if (s >= CANCELLED) throw new CancellationException(); throw new ExecutionException((Throwable)x); }
FutureTask#finishCompletion()
方法用来唤醒前面等待的线程
上一步awaitDone
方法会阻塞调用的线程,那么任务运行结束总要唤醒他们去拿结果吧,这个工作就在finishCompletion()
方法中。
private void finishCompletion() { // 遍历waiters栈中的每个元素; for (WaitNode q; (q = waiters) != null;) { // cas设置waiters中q节点数据为null,成功的话,进入到if中 if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) { // 自选操作 for (;;) { // 获取节点中的线程 Thread t = q.thread; if (t != null) { q.thread = null; // 唤醒线程 LockSupport.unpark(t); } // 获取下一个节点 WaitNode next = q.next; if (next == null) break; q.next = null; // unlink to help gc q = next; } break; } } //钩子方法,有子类去实现 done(); // 设置原来的任务callable为null callable = null; // to reduce footprint }
任务取消cancel()原理
可以调用FutureTask#cancel
方法取消任务执行,但是要注意下面几点:
- 任务取消时会先检查是否允许取消,当任务已经完成或者正在完成(正常执行并继续处理结果 或 执行异常处理异常结果)时不允许取消。
- cancel方法有个boolean入参,若为false,则只唤醒所有等待的线程,不中断正在执行的任务线程。若为true则直接中断任务执行线程,同时修改状态为INTERRUPTED。
// 取消任务,参数mayInterruptIfRunning为true,会中断运行中的线程,false不会 public boolean cancel(boolean mayInterruptIfRunning) { // 如果FutureTask的状态不是NEW或者CAS设置失败时,直接返回false if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { // 如果参数mayInterruptIfRunning为true,中断 if (mayInterruptIfRunning) { try { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state //cas修改状态为INTERRUPTED UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { // 唤醒其他等待的线程 finishCompletion(); } return true; }
cancel方法实际上完成以下两种状态转换之一:
- NEW -> CANCELLED (对应于mayInterruptIfRunning=false)
- NEW -> INTERRUPTING -> INTERRUPTED (对应于mayInterruptIfRunning=true)
总结
本文讲解了线程池submit()提交任务的原理实现,通过源码很多平时项目中遇到的坑都找到了答案。比如说之前项目中用线程池submit()方法提交任务处理,发现任务的异常都不见了,这下明白了,原来是通过setException()保存下来了,只有通过get方法获取到,只有看过源码,才会豁然开朗。
以上就是Java线程池submit阻塞获取结果的实现原理详解的详细内容,更多关于Java线程池submit阻塞获取结果的资料请关注IT俱乐部其它相关文章!