JDK future框架,提供了一种异步编程模式,基于线程池的。将任务runnable/callable提交到线程池executor,返回一个Future对象。通过future.get()获取执行结果,这里提交到线程池,后面的操作不会阻塞。future.get()获取结果会阻塞,其实也是用多线线程执行任务。
future.get()这里会阻塞,google的guava提供了一个calllback解决办法,这也是我准备看的
下面是一个future的demo
import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; /** * @Author: <guanxianseng@163.com> * @Description: jdk future demo * @Date: Created in : 2018/11/30 6:01 PM **/ public class TestJdkFuture { public static void main(String[] args) throws ExecutionException, InterruptedException { testJdkFuture(); } public static void testJdkFuture() throws ExecutionException, InterruptedException { Callable<Integer> callable = () ->{ System.out.println("callable do some compute"); return 1; } ; Runnable runnable = () -> { System.out.println("runable do some compute"); }; ExecutorService executorService = Executors.newCachedThreadPool(); Future<Integer> future = executorService.submit(callable); Future runableFuture = executorService.submit(runnable); Object runableRes = runableFuture.get(); int res = future.get(); executorService.shutdown(); System.out.println("callable res: " + res); System.out.println("runnable res: " + runableRes); } }
demo里面提交了一个Callable和一个Runnable到线程池,通过future获取计算结果
executorService.submit(callable)源码
public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
这里主要是封装了一个RunnableFuture和提交任务到线程池
我们看下RunnableFuture里面的run方法,因为线程池执行任务,是执行run方法。看FutureTask里面的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 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 interrupts int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
首先,设置runner为线程池中的当前线程,后面执行call()方法,计算出结果,set(result)。跟进set(result)
/** * Sets the result of this future to the given value unless * this future has already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon successful completion of the computation. * * @param v the value */ protected void set(V v) { if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); } }
这里用了一个cas设置FutureTask的state字段为COMPLETING,完成中的一个状态。接着设置outcom为计算结果,我们跟进finishCompletion()
/** * Removes and signals all waiting threads, invokes done(), and * nulls out 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 }
WaitNode主要是一个阻塞线程链表,即调用future.get()方法的线程链表。这里主要作用是注意唤醒这些线程,通过LockSupport.unpark(t)唤醒。这里用阻塞线程链表,主要是考虑到可能有多个线程会调用future.get()阻塞
ok,到这里执行任务,把计算结果放到future中,并唤醒阻塞线程已经理清楚了
我们再来看下,future.get()是如何实现阻塞,和获取到计算结果的
进入future.get()
/** * Waits if necessary for the computation to complete, and then * retrieves its result. * * @return the computed result * @throws CancellationException if the computation was cancelled * @throws ExecutionException if the computation threw an * exception * @throws InterruptedException if the current thread was interrupted * while waiting */ V get() throws InterruptedException, ExecutionException;
Future.java是一个接口,这里看注释可以看出,会阻塞等待结果计算完成。我们看下一个实现FutureTask.java的get()方法
/** * @throws CancellationException {@inheritDoc} */ public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L); return report(s); }
awaitDone(false, 0l)主要是阻塞线程,进入方法
/** * Awaits completion or aborts on interrupt or timeout. * * @param timed true if use timed waits * @param nanos time to wait, if timed * @return state upon completion */ 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); } }
这里,主要是阻塞线程,把当前线程放到阻塞线程链表中,通过LockSupport.park(this)阻塞当前线程,等待线程池里面的线程唤醒。唤醒之后,回到get()方法,看report()方法
/** * Returns result or throws exception for completed task. * * @param s completed state value */ @SuppressWarnings("unchecked") 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); }
这里主要是返回计算结果给阻塞线程
到这里,基本理清楚了,future的阻塞实现,以及获取计算结果的步骤。
future框架 = 线程池 + futrue