java同步和异步的举例,java异步代码

　　00-1010了解什么是FutureTask？FutureTask是如何实现的？FutureTask使用前言：运行进程FutureTask

　　你熟悉FutureTask吗？还是有异步计算的需求？FutureTask可以帮助你实现异步计算，可以同步获取异步任务的计算结果。我们从源代码来分析一下FutureTask。

　　包Java . util . concurrent；Interface Futurev {/* * *试图取消任务的执行并返回取消结果。*参数mayInterruptIfRunning:是否中断线程。*/boolean cancel(boolean mayInterruptIfRunning)；/* * *确定任务是否已经取消(在正常结束前取消并返回true)*/boolean is cancelled()；/* * *判断当前任务是否已经执行，包括正常执行、异常执行或任务取消。*/布尔isDone()；/* * *获取任务执行结果，任务结束前会阻塞。*/V get()抛出InterruptedException，ExecutionException/* * *尽量在规定时间内得到执行结果。如果超时，超时异常timeout exception */v get(长超时，时间单位单位)抛出中断异常、执行异常、超时异常；}Runnable接口我们都很熟悉。它是一个函数接口，我们经常用它来创建一个线程。

　　包java.lang@ functional interface public interface Runnable { public abstract void run()；}FutureTask是一个要执行的任务，包含了上述接口的具体实现。FutureTask内部定义了任务的状态state和一些状态常量，其内部核心是可调用Callable。我们可以通过构造函数传入callable或者runnable，最终会在内部转换成callable，因为我们需要获取异步任务的执行结果，只有通过callable创建的线程才会返回结果。

　　我们可以通过此时的状态来判断isCancelled()和isDone()未来的返回结果。

　　以下为FutureTask源码，内含核心源码分析注释

　　包Java . util . concurrent；导入Java . util . concurrent . locks . locksupport；公共类FutureTaskV实现

　　s RunnableFuture<V> { /** * 任务的运行状态 */ 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; // 中断的 private Callable<V> callable; /** * 返回结果 */ private Object outcome; private volatile Thread runner; private volatile WaitNode waiters; ... public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; this.state = NEW; } public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; } public boolean isCancelled() { return state >= CANCELLED; } public boolean isDone() { return state != NEW; } /* * 取消任务实现 * 如果任务还没有启动就调用了cancel(true)，任务将永远不会被执行。 * 如果任务已经启动，参数mayInterruptIfRunning将决定任务是否应该中断执行该任务的线程，以尝试中断该任务。 * 如果任务任务已经取消、已经完成或者其他原因不能取消，尝试将失败。 */ public boolean cancel(boolean mayInterruptIfRunning) { if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { // in case call to interrupt throws exception if (mayInterruptIfRunning) { try { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { finishCompletion(); } return true; } /* * 等待获取结果 * 获取当前状态，判断是否执行完成。并且判断时间是否超时 * 如果任务没有执行完成，就阻塞等待完成，若超时抛出超时等待异常。 */ public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L); return report(s); } /* * 等待获取结果 * 获取当前状态，判断是否执行完成。 * 如果任务没有执行完成，就阻塞等待完成。 */ 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; if (s == NORMAL) return (V)x; if (s >= CANCELLED) throw new CancellationException(); throw new ExecutionException((Throwable)x); } protected void done() { } /** * 设置结果借助CAS确认状态是否完成状态 */ protected void set(V v) { if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); } } /** * 设置异常，当运行完成出现异常，设置异常状态 */ protected void setException(Throwable t) { if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = t; UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state finishCompletion(); } } /* * 执行callable获取结果，或者异常 * 判断状态是不是启动过的，如果是新建才可以执行run方法 */ public void run() { if (state != NEW !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { 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; setException(ex); } if (ran) set(result); } } finally { runner = null; int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } } /** * 重新执行 */ protected boolean runAndReset() { if (state != NEW !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return false; boolean ran = false; int s = state; try { Callable<V> c = callable; if (c != null && s == NEW) { try { c.call(); // dont set result ran = true; } catch (Throwable ex) { setException(ex); } } } finally { runner = null; s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } return ran && s == NEW; } /* * 处理可能取消的中断 */ private void handlePossibleCancellationInterrupt(int s) { if (s == INTERRUPTING) while (state == INTERRUPTING) Thread.yield(); } static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread.currentThread(); } } /** * 移除并唤醒所有等待线程,执行done,置空callable */ private void finishCompletion() { // assert state > COMPLETING; for (WaitNode q; (q = waiters) != null;) { 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; // to reduce footprint } /** * 等待完成 * 首先判断是否超时 * 处理中断的，然后处理异常状态的，处理完成的... */ 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); } } /** * 去除等待 */ private void removeWaiter(WaitNode node) { if (node != null) { node.thread = null; retry: for (;;) { // restart on removeWaiter race for (WaitNode pred = null, q = waiters, s; q != null; q = s) { s = q.next; if (q.thread != null) pred = q; else if (pred != null) { pred.next = s; if (pred.thread == null) // check for race continue retry; } else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s)) continue retry; } break; } } } // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE; private static final long stateOffset; private static final long runnerOffset; private static final long waitersOffset; static { try { UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> k = FutureTask.class; stateOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("state")); runnerOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("runner")); waitersOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("waiters")); } catch (Exception e) { throw new Error(e); } }}

FutureTask 运行流程

一般来说，我们可以认为FutureTask具有以下三种状态：

　　未启动：新建的FutureTask，在run()没执行之前，FutureTask处于未启动状态。

private static final int NEW = 0; // 新建

已启动：FutureTask对象的run方法启动并执行的过程中，FutureTask处于已启动状态。

　　已完成：FutureTask正常执行结束，或者FutureTask执行被取消(FutureTask对象cancel方法)，或者FutureTask对象run方法执行抛出异常而导致中断而结束，FutureTask都处于已完成状态。

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; // 中断的

FutureTask 的使用

使用一（直接新建一个线程调用）：

FutureTask<Integer> task = new FutureTask<>(new Callable() {@Override public Integer call() throws Exception { return sum(); }});new Thread(task).stat();Integer result = task.get();

使用二（结合线程池使用）

FutureTask<Integer> task = new FutureTask<>(new Callable() {@Override public Integer call() throws Exception { return sum(); }});Executors.newCachedThreadPool().submit(task);Integer result = task.get();

到此这篇关于Java从源码看异步任务计算FutureTask的文章就介绍到这了,更多相关Java FutureTask内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

java同步和异步的举例,java异步代码

目录

FutureTask 运行流程

FutureTask 的使用

相关文章阅读