FutureTask为何单个任务仅执行一次原理解析
作者:zhongh Jim
引言
前几天会员领取情况查询的接口SQL查询超时出故障了,因为有个用户买的会员有点多(哈哈),其实是 数据量大 + 祖传代码逻辑冗长
尝试的解决方案:
SQL:检查了一下,单个SQL的耗时其实不算大,也能接受,不需要改动,主要原因是后端逻辑冗长
FutureTask获取线程的执行结果:将1次大查询划分为多次小查询同时进行,提高接口响应速度。且一个FutureTask仅执行一次,不会出现重复的查询
经过权衡,我们选择了后者
一、FutureTask用法
解决方案要用到线程池搭配FutureTask,这里我们就不用了,简化点
public class Test { //计算结果 int count=0; @Test public void test(){ try{ FutureTask<Integer> futureTask=new FutureTask<>(new Callable<Integer>() { @Override public Integer call() throws Exception { return 1; } }); //把FutureTask放入线程中,线程会运行FutureTask的run()代码块 Thread t1=new Thread(futureTask); t1.start(); //获取计算的结果,是一个阻塞等待返回的方法 count+=futureTask.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } //最后结果: 1 System.out.println(count); } }
这里用了构造方法public FutureTask(Callable<V> callable)让FutureTask持有Callable接口的实例
用到try-catch是由于futureTask.get()方法是一个阻塞等待的过程,途中如果被中断会抛中断异常,别的异常都会以ExecutionException执行异常的形式抛出
二、(重要)FutureTask的任务仅执行一次,为何?
FutureTask的run()代码块仅执行一次!请看注释
/** 执行结果(全局变量), 有2种情况: 1. 顺利完成返回的结果 2. 执行run()代码块过程中抛出的异常 */ private Object outcome; //正在执行run()的线程, 内存可被其他线程可见 private volatile Thread runner; public void run() { /** FutureTask的run()仅执行一次的原因: 1. state != NEW表示任务正在被执行或已经完成, 直接return 2. 若state==NEW, 则尝试CAS将当前线程 设置为执行run()的线程,如果失败,说明已经有其他线程 先行一步执行了run(),则当前线程return退出 */ if (state != NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())) return; try { //持有Callable的实例,后续会执行该实例的call()方法 Callable<V> c = callable; if (c != null && state == NEW) { V result; boolean ran; try { result = c.call(); ran = true; }catch (Throwable ex) { result = null; ran = false; //执行中抛的异常会放入outcome中保存 setException(ex); } if (ran) //若无异常, 顺利完成的执行结果会放入outcome保存 set(result); } }finally { // help GC runner = null; int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
执行run()的代码块之后,其他线程如何拿到FutureTask的执行结果?下面的get()方法可以做到
三、get()获取结果
public V get() throws InterruptedException, ExecutionException { int s = state; //COMPLETING: 正在完成的状态; s <= COMPLETING就是未完成 if (s <= COMPLETING) //不计时等待,结束等待的条件只有【完成】、【被中断】、【被取消】、【抛其他异常(不包括中断异常、取消异常)】 s = awaitDone(false, 0L); return report(s); }
这里提一下线程执行的状态 :
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;
volatile保证了线程执行的状态改变之后会刷新到内存中,被其他线程可见
如果线程还处于未完成的状态,即s <= COMPLETING,就会进入等待状态,调用awaitDone(false, 0L)方法
get为何阻塞等待?
/** @param timed 若是true则为定时等待,超时后会结束等待,并返回当前状态state @param nanos 如果是定时等待即第一个入参timed=true的话,会设置对应的等待时长 */ private int awaitDone(boolean timed, long nanos) throws InterruptedException { //等待的最后期限 final long deadline = timed ? System.nanoTime() + nanos : 0L; WaitNode q = null; boolean queued = false; //进入无限循环的等待状态,只有【完成】、【被取消】、【异常】、【中断】、【超时】这五种情况才会结束等待 for (;;) { if (Thread.interrupted()) { //线程执行被中断,则移除等待结点并抛出异常 removeWaiter(q); throw new InterruptedException(); } int s = state; //【完成】、【被取消】、【抛其他异常】的状态都会 在这 结束等待 if (s > COMPLETING) { if (q != null) q.thread = null; return s; } //子线程处于任务完成的瞬时状态,要等一会才能拿到执行结果 else if (s == COMPLETING) // cannot time out yet Thread.yield(); else if (q == null) q = new WaitNode(); else if (!queued) queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); else if (timed) { //设置定时等待并且已经超时了 nanos = deadline - System.nanoTime(); if (nanos <= 0L) { removeWaiter(q); return state; } LockSupport.parkNanos(this, nanos); } else LockSupport.park(this); } }
详细的注释在代码中,请耐心看一下。
简单来说,能结束等待的条件只有5个:
- 完成
- 被中断
- 设置定时等待并超时
- 被取消
- 抛了其他异常,比如RuntimeException,这里的其他异常既不是中断异常,也不是取消异常
调用futureTask.get()的等待方式有2种,分为定时等待和 不计时等待:
- timed=true是定时等待,会创建等待结点q = new WaitNode();并放在栈顶(队列头部),然后挂起。结束等待的条件(满足任一即可)是【完成】、【被中断】、【被取消】、【抛其他异常】、【超时】 。
- timed=false是不计时等待,创建等待结点后会一直挂起,只有【完成】、【被中断】、【被取消】、【抛其他异常】
在等待结束之前,LockSupport.park(this);表示线程会被一直挂起,不再继续无限循环占用CPU。
解除挂起的条件是state > COMPLETING,然后调用finishCompletion()方法去让线程解除挂起并回到awaitDone()做最后一次循环后return state
从get中返回结果report(int s)
/*正常的计算结果 or 抛出的异常 都会作为outcome*/ private Object outcome; private V report(int s) throws ExecutionException { Object x = outcome; //正常完成 if (s == NORMAL) return (V)x; //执行的过程中【被取消】 if (s >= CANCELLED) throw new CancellationException(); /** 这里抛的是执行过程中发生的其他异常,既不是【中断异常】,也不是【被取消异常】 比如发生了RuntimeException之类的就会在这抛 */ throw new ExecutionException((Throwable)x); }
report(int s)是执行get()获取结果的最后一步
看到这可能有朋友晕了,我把get()内部的流程梳理一下:
若要等待计算结果:get() -> awaitDone() -> report(),共3步
不用等待:get() -> report() ,仅2步
四、FutureTask是如何拿到线程执行的结果?
主要 有赖于FutureTask类内部的Callable接口
只有Callable接口能拿到线程的返回值,下面来看下FutureTask的构造函数
public class FutureTask<V> implements RunnableFuture<V> { //执行任务并返回结果 private Callable<V> callable; public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; //新建状态 this.state = NEW; } }
其实Callable 接口是没法 作为创建线程new Thread(Runnable target)的入参的,只有借助FutureTask类才能被线程执行,因为FutureTask实现了Runnable 接口
有兴趣的可以看一下Future接口的关系图(这里拿了大佬的图,侵删)
FutureTask类最终实现了Future接口和Runnable接口,可作为new Thread(Runnable target)的入参target来创建线程
五、FutureTask可能的执行过程
顺利完成 :NEW -> COMPLETING -> NORMAL ,即新建->正在完成 ->正常
NEW -> COMPLETING -> EXCEPTIONAL, 执行过程出现了异常
被取消:NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED,新建 ->正在被中断 ->中断完成
六、列举一下FutureTask的特性和应用场景
特性:
- 异步执行,可执行多次(通过runAndReset()方法),也可仅执行一次(执行run()即可)
- 可获取线程执行结果
应用场景:
- 长时间运行的任务,包含远程调用的任务
- 数据量大的查询,划分为多个小查询,每个FutureTask 仅执行一次 的特性能有效避免重复的查询
- 计算密集型的任务
以上就是FutureTask为何单个任务仅执行一次原理解析的详细内容,更多关于FutureTask单任务执行一次的资料请关注脚本之家其它相关文章!