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: <[email protected]>
* @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
原文地址：https://www.cnblogs.com/luckygxf/p/10048238.html
时间： 2024-10-17 08:44:08