赞
踩
Future,FutureCallback,promise都是多线程编程中,线程间同步的方式。
1.jdk的future,jdk 1.8
与futureTask,Callable 等数据结构用,用于获取异步执行的线程的返回结果。
java创建线程的方式只有Thread,runnable接口只是给Thread类提供执行的任务和业务逻辑,Thread的start方法才启动一个线程来执行。 callable方法不是直接被thread类使用的,是和
futureTask一起使用,futureTask中的run方法中调用Callable类的call方法,获取返回值.
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
public class FutureTask<V> implements RunnableFuture<V> {
/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
...
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);
}
}
future 用于获取线程执行结果,通常情况下,get方法会阻塞等到线程执行完毕,最底层的机制还是通过互斥锁+条件变量,来实现等待通知机制,这样避免自旋和轮询,长期占用cpu.
/**
* @throws CancellationException {@inheritDoc}
*/
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);
}
awaitDone方法实际执行等待操作。如果已经执行完毕,直接返回状态,如果是COMPLETING,放弃cpu,等待下一次调度,如果是其他状态,最终毁掉用unsafe.park函数阻塞,等待unpark函数的唤醒。
/**
* 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);
}
}
futureTask的run函数在执行完成call函数后,会调用finishCompletion函数,来唤醒等待在future的get方法上的线程
/**
* 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
}
2.guava的FutureCallback
FutureCallback 可以在任务执行完毕后,自动调用注册的回调函数,不用再其他线程get结果后再进行处理。
Callback的实现原理可以的想到在对应的Future
中维护一个Callback
链表,当任务执行完成后依次执行对应的回调,类似于观察者模式的Subject
依次调用Observer
。使用callback机制可能存在一个嵌套的问题,俗称回调地狱。
public static <V> void addCallback(final ListenableFuture<V> future,
final FutureCallback<? super V> callback, Executor executor) {
Preconditions.checkNotNull(callback);
Runnable callbackListener = new Runnable() {
@Override
public void run() {
final V value;
try {
// TODO(user): (Before Guava release), validate that this
// is the thing for IE.
value = getUninterruptibly(future);
} catch (ExecutionException e) {
callback.onFailure(e.getCause());
return;
} catch (RuntimeException e) {
callback.onFailure(e);
return;
} catch (Error e) {
callback.onFailure(e);
return;
}
callback.onSuccess(value);
}
};
future.addListener(callbackListener, executor);
}
@Override
public void addListener(Runnable listener, Executor exec) {
executionList.add(listener, exec);
}
3,promise
es6 和 netty都有promise机制。promise 机制提供在一个任务执行完毕后,自动执行另外一个任务的功能。ES6规定,Promise 对象是一个构造函数,用来生成Promise 实例。Promise构造函数接受一个函数作为参数,该函数有两个参数分别是resolve和reject,它们也是函数。resolve函数的作用是,将Promise 对象的状态从“未完成”(pending)==>“成功”(resolved),在异步操作成功时调用,并将异步操作结果,作为参数传递出去。reject函数的作用是,将Promise 对象的状态从“未完成”(pending)==>“失败”(rejected),再异步操作失败时调用,并将操作报错的错误,作为参数传递出去。Promise
实例生成以后,可以用then
方法分别指定resolved
状态和rejected
状态的回调函数,如下:
promise.then(function(value) {
// success
}, function(error) {
// failure
});
netty的promise, 除了增加异步任务执行结束,自动回调注册的函数外,还增加了
可写 的api,用于主动设置异步任务的执行状态。具有设置成功结果,设置失败结果等功能,这样可以在成功或失败的时候回调注册到当前Promise实例的listeners了。在业务逻辑执行前添加监听器addListener(FutureListener<V> listener)
,在执行完业务逻辑之后,执行setSuccess/trySuccess/setFailure/tryFailure
等方法,此时会执行notifyAll()并回调添加进来的监听器。假设有线程阻塞在get()方法上时,在此时会做唤醒。
4.jdk8的CompletableFuture
Future和CompletableFuture一个是接口一个是其实现类。
CompletableFuture 也是提供异步任务执行完毕后,自动回调已注册的函数的功能,不用在主线程等待获取执行结果。CompletableFuture
类,实现了Promise模式编程,除了回调功能外,也有主动完成任务的能力,即使异步任务会导致异步线程无限休眠,但是仍然可以通过主动设置值的方式完成该任务,是通过complete函数来完成的。下面是网上的一个例子:
CompletableFuture<String> promise = CompletableFuture.supplyAsync(() -> { // 无限休眠 sleep(Long.MAX_VALUE); return null; }); // 调用主动完成 promise.complete("quding2"); System.out.println(promise.get()); // 输出quding2
用以下函数来创建异步执行任务:
runAsync(Runnable runnable) 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码。
runAsync(Runnable runnable, Executor executor) 使用指定的thread pool执行异步代码。
supplyAsync(Supplier<U> supplier) 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码,异步操作有返回值
supplyAsync(Supplier<U> supplier, Executor executor) 使用指定的thread pool执行异步代码,异步操作有返回值
CompletableFuture还提供了注册回调函数,异常处理,等待最快一个或者全部异步任务完成。
通过whenXXX来设置回调函数,在一个任务执行完成之后调用的方法。这个有三个名差不多的方法。whenComplete、whenCompleteAsync、还有一个是whenCompleteAsync用自定义Executor。
网上的一个例子,如何设置回调:
public static void asyncCallback() throws ExecutionException, InterruptedException { CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() { @Override public String get() { System.out.println(getThreadName()+"supplyAsync"); return "123"; } }); CompletableFuture<Integer> result1 = task.thenApply(number->{ System.out.println(getThreadName()+"thenApply1"); return Integer.parseInt(number); }); CompletableFuture<Integer> result2 = result1.thenApply(number->{ System.out.println(getThreadName()+"thenApply2"); return number*2; }); System.out.println(getThreadName()+" => "+result2.get()); }
5.ForkJoinPool.commonPool()
runAsync和supplyAsync 在指定线程池时,用ForkJoinPool.commonPool()来作为执行任务的线程池。
ForkJoinPool是ExecutorSerice的一个补充,而不是替代品,从JDK1.7开始,Java提供ForkJoin框架用于并行执行任务,它的思想就是讲一个大任务分割成若干小任务,最终汇总每个小任务的结果得到这个大任务的结果。ForkJoinPool 主要用于实现“分而治之”的算法,特别是分治之后递归调用的函数,例如 quick sort 等。ForkJoinPool 最适合的是计算密集型的任务,如果存在 I/O,线程间同步,sleep() 等会造成线程长时间阻塞的情况时,最好配合使用 ManagedBlocker。
ForkJoinPool 是一个线程池,ForkJoinTask就是ForkJoinPool里面的每一个任务,主要有两个子类:RecursiveAction和RecursiveTask。然后通过fork()方法去分配任务执行任务,通过join()方法汇总任务结果,
ForkJoinPool.commonPool()是一个全局的线程池,和jvm的进程生命周期一样,如果显示创建线程池来执行任务,runAsync和supplyAsync会使用ForkJoinPool.commonPool(),引入commonPool,是为了避免任何并行操作都引入一个线程池,最坏情况会导致在单个JVM上创建了太多的池线程,降低效率。但是在 JVM 的后台,使用通用的 fork/join 池来完成一些功能时,该池是所有并行流共享的。默认情况,fork/join 池会为每个处理器分配一个线程。假设你有一台16核的机器,这样你就只能创建16个线程。对 CPU 密集型的任务来说,这样是有意义的,因为你的机器确实只能执行16个线程。但是真实情况下,不是所有的任务都是 CPU 密集型的。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。