免费一级域名注册网站,php酒店网站源码,最新新闻热点事件2023,wordpress显示关闭评论框AbstractExecutorService
上一篇文章中#xff0c;主要介绍了AbstractExecutorService的线程执行的核心流程#xff0c;execute() 这个方法显然是没有返回执行任务的结果#xff0c;如果我们需要获取任务执行的结果#xff0c;怎么办#xff1f;
Callable 就是一个可以获…AbstractExecutorService
上一篇文章中主要介绍了AbstractExecutorService的线程执行的核心流程execute() 这个方法显然是没有返回执行任务的结果如果我们需要获取任务执行的结果怎么办
Callable 就是一个可以获取线程执行的结果。
public abstract class AbstractExecutorService implements ExecutorService {/** 将任务包装成FutureTask任务。带返回值参数的*/protected T RunnableFutureT newTaskFor(Runnable runnable, T value) {return new FutureTaskT(runnable, value);}/**** 不带返回值的**/protected T RunnableFutureT newTaskFor(CallableT callable) {return new FutureTaskT(callable);}/*** throws RejectedExecutionException {inheritDoc}* throws NullPointerException {inheritDoc}*/public Future? submit(Runnable task) {if (task null) throw new NullPointerException();//1.将任务包装成RunableFuture对象由于RunnableFuture是实现Runable类所以execute的参数是一个可拓展的类型RunnableFutureVoid ftask newTaskFor(task, null);//2交给具体的执行器进行实现execute(ftask);return ftask;}/*** throws RejectedExecutionException {inheritDoc}* throws NullPointerException {inheritDoc}*/public T FutureT submit(Runnable task, T result) {if (task null) throw new NullPointerException();RunnableFutureT ftask newTaskFor(task, result);execute(ftask);return ftask;}/*** throws RejectedExecutionException {inheritDoc}* throws NullPointerException {inheritDoc}*/public T FutureT submit(CallableT task) {if (task null) throw new NullPointerException();//将任务装成成一个FutureTask任务RunnableFutureT ftask newTaskFor(task);//执行任务execute(ftask);return ftask;}}submit其实是一个重载的方法分别是一个task以及可以传递获取结果的任务以及使用callable。
demo
从源码上看三个方法其实都是将任务进行了封装然后调用线程池执行的核心方法 public static void main(String[] args) throws ExecutionException, InterruptedException {CallableInteger resultCallable new CallableInteger() {Overridepublic Integer call() throws Exception {return 1 1;}};ExecutorService threadPool Executors.newFixedThreadPool(1);FutureInteger resultTask threadPool.submit(resultCallable);System.out.println(resultTask.get());threadPool.shutdown();}FutureTask
public class FutureTaskV implements RunnableFutureV {/* NEW - COMPLETING - NORMAL* NEW - COMPLETING - EXCEPTIONAL* NEW - CANCELLED* NEW - INTERRUPTING - INTERRUPTED*/private volatile int state;private static final int NEW 0; // 初始化状态private static final int COMPLETING 1; // 结果计算完成或响应中断到赋值给返回值的状态private static final int NORMAL 2; // 任务正常完成结果被setprivate static final int EXCEPTIONAL 3; // 任务抛出异常private static final int CANCELLED 4; // 任务被取消private static final int INTERRUPTING 5; // 线程中断状态被设置为true 线程未响应中断private static final int INTERRUPTED 6; // 线程已被中断/** The underlying callable; nulled out after running */private CallableV callable; // 需要执行的任务/** The result to return or exception to throw from get() */// 执行callable的线程调用FutureTask.run()方法通过CAS设置private Object outcome; // non-volatile, protected by state reads/writes/** The thread running the callable; CASed during run() */// 执行callable的线程调用FutureTask.run()方法通过CAS设置private volatile Thread runner;/** Treiber stack of waiting threads */private volatile WaitNode waiters;public FutureTask(CallableV callable) {if (callable null)throw new NullPointerException();this.callable callable;this.state NEW; // 初始化状态是new // ensure visibility of callable}
}/* 继承了Runnable 因为线程池中执行的也是Runnbale的任务*/
public interface RunnableFutureV extends Runnable, FutureV {/*** Sets this Future to the result of its computation* unless it has been cancelled.*/void run();
}FutureTask 实现RunnableFuture也间接实现了run方法。
重点
我们知道 execute(ftask); 本质就是利用线程池进行执行而线程执行的时候其实就是启动对应任务的run方法。
task.run();// 这里是什么时候调用的其实是// execute(ftask)传入的任务 task.run()public void run() {//不是新建状态 直接中止if (state ! NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {CallableV c callable;if (c ! null state NEW) {V result;boolean ran;try {//核心执行任务的call方法你看就是调用普通的方法一样。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;// state must be re-read after nulling runner to prevent// leaked interruptsint s state;//响应中断if (s INTERRUPTING)handlePossibleCancellationInterrupt(s);}}判断当前任务状态非NEW直接返回执行对应c.call() 其实就是执行callable中的call方法。将返回值set进去 protected void set(V v) {//CAS 去设置当前任务执行状态 new-completingif (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {//返回结果outcomeoutcome v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}}get public V get() throws InterruptedException, ExecutionException {int s state;//如果是在执行中则等待一会if (s COMPLETING)s awaitDone(false, 0L);//返回结果return report(s);}/*** throws CancellationException {inheritDoc}*/public V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {if (unit null)throw new NullPointerException();//设置了超时时间则等待一定的时间如果还没有获取到返回异常int s state;if (s COMPLETING (s awaitDone(true, unit.toNanos(timeout))) COMPLETING)throw new TimeoutException();return report(s);}private V report(int s) throws ExecutionException {Object x outcome;//执行完成 返回x结果if (s NORMAL)return (V)x;//如果任务取消返回异常if (s CANCELLED)throw new CancellationException();throw new ExecutionException((Throwable)x);}awaitDone private int awaitDone(boolean timed, long nanos)throws InterruptedException {final long deadline timed ? System.nanoTime() nanos : 0L;WaitNode q null;boolean queued false;for (;;) {//如果线程执行interrupted直接抛出异常并且将任务移除if (Thread.interrupted()) {removeWaiter(q);throw new InterruptedException();}int s state;//状态大于COMPLETING 说明完成了if (s COMPLETING) {if (q ! null)q.thread null;return s;}//else if (s COMPLETING) // cannot time out yetThread.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);}elseLockSupport.park(this);}}小结
FutureTask是一个支持取消行为的异步任务执行器。该类实现了Future接口的方法。 如
取消任务执行查询任务是否执行完成获取任务执行结果”get“任务必须得执行完成才能获取结果否则会阻塞直至任务完成。
如果在当前线程中需要执行比较耗时的操作但又不想阻塞当前线程时可以把这些作业交给FutureTask另开一个线程在后台完成当当前线程将来需要时就可以通过FutureTask对象获得后台作业的计算结果或者执行状态。
Future模式其实是多线程编程中常用的设计模式主线程向另外一个线程提交任务无需等待任务执行的结果返回一个凭证就是future通过future.get()去获取结果。这个过程可能是阻塞的。
