本文源码版本1.8.0_241,更高版本源码会有所不同。
常量
CompletableFuture类里只有2个常量
volatile Object result;// Either the result or boxed AltResult
volatile Completion stack;// Top of Treiber stack of dependent actions
result是用来保存任务执行的结果,stack是用来保存阻塞等待结果的线程封装对象,
abstract static class Completion extends ForkJoinTask implements Runnable, AsynchronousCompletionTask {
volatile Completion next;// Treiber stack link
/**
* Performs completion action if triggered, returning a
* dependent that may need propagation, if one exists.
*
* @param mode SYNC, ASYNC, or NESTED
*/
abstract CompletableFuture tryFire(int mode);
/** Returns true if possibly still triggerable. Used by cleanStack. */
abstract boolean isLive();
public final void run() { tryFire(ASYNC); }
public final boolean exec() { tryFire(ASYNC); return true; }
public final Void getRawResult() {return null; }
public final void setRawResult(Void v) {}
}
从名字可以看出,这些等待结果的线程封装对象是以栈结构保存的,后文会说到。2个变量都是以volatile修饰的,保证可见性。
带返回结果的执行方法
CompletableFuture中提供了2种执行方法,一种没有返回结果,另一种有返回结果,先看有返回结果的
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
return asyncSupplyStage(asyncPool, supplier);
}
以supplyAsync为例,进到asyncSupplyStage方法
创建了一个CompletableFuture对象并作为参数传给了AsyncSupply对象,很明显,AsyncSupply对象是一个封装的任务类,
里面有2个变量,一个CompletableFuture用来保存运行结果,一个Supplier对象,函数式接口,保存执行的方法,内部有个run方法用来执行方法,核心就是d.completeValue(f.get()); 进去completeValue看看
这里使用到了大名鼎鼎的UNSAFE对象进行cas操作,cas方法有4个参数,分别是对象,变量在内存上在对象内部的相对偏移地址,期望值,更新值,这里先说下这个RESULT
他是在CompletableFuture初始化时,通过反射获取到result变量在CompletableFuture对象里的内存地址偏移量,然后保存了起来;所以上面cas操作的意思是如果RESULT位置处的result变量为null,那么就更新值为后面的三元判断表达式得出的值,而result变量默认值是为null的,所以如果运行的方法结果为null,那么返回NIL对象,NIL是什么
可以看到是一个封装了null的对象,所以如果执行的方法结果是null,那么返回一个封装了null值的对象,否则返回执行的结果。
回到一开始的run方法,可以看到在方法执行结束后执行了一个d.postComplete()方法
如果该方法执行比较耗时,而在其他地方有线程阻塞等待方法执行的结果,那么在这里会将那些阻塞等待的方法唤醒,可以看到while循环里的表达式,(h = f.stack) !=null,如果有线程阻塞等待的话,这个h是不为null的,那么会进入到循环里,在方法的最后执行tryFire方法,
可以看到这个方法是将阻塞等待的线程唤醒,如果此时有多个线程阻塞等待的话,那么通过if条件里的f.casStack(h, t = h.next)这个cas操作,会依次将所有线程唤醒。
另外,该方法还有一个同名方法,有2个参数,支持传入自定义的线程池
另外,值得注意的一点是,如果不传入自定义线程池的话,使用的是底层自带的ForkJoinPool池,该池的线程均为守护线程,所以使用时需要注意,可能出现线程池里的线程还没执行完,外面的主线程已经执行结束了,可以调用ForkJoinPool的awaitQuiescence方法等待线程池里的线程执行结束。
不带返回结果的执行方法
以runAsync方法为例
同样的套路,进入asyncRunStage方法
一样的封装了个AsyncRun对象
可以看到大体的结构和之前是一样的,唯一的不同在于,在方法执行结束后调用了d.completeNull()这个方法,
这个cas操作判断result变量是否为null,为null就更新为之前说的那个封装了null值的对象,而result变量默认是为null的,所以直接返回一个封装了null值的对象,因为run方法是没有返回值的,所以直接就这样返回了。其他的逻辑和之前说的一样。
get方法
接下来看看get方法
如果调用get方法时,方法还未执行完成,那么此时result是为null的,调用waitingGet方法
可以看到会进入一个while循环里,前2个if是自旋一段非常短的时间,在更高版本的源码中,这2个if被删除了,此处有个有意思的东西,可以看到有一个SPINS常量,这个常量是什么呢,
他调用了系统函数判断机器是否是多核CPU,如果是则赋值256,否则为0,这个变量在之前的版本里是写在方法里的,但是会有性能问题,所以后来抽出来了,有兴趣的可以看看https://bugs.openjdk.java.net/projects/JDK/issues/,我们接着往下看,当前2个if执行完后,执行到第三个if,会构建一个Signaller对象
可以看到里面有一个thread变量保存了阻塞等待的线程,前面说的tryFire方法唤醒线程也在这里,接着走到第四个if,执行tryPushStack方法
这里就是把所有阻塞等待的线程用栈保存起来,怎么做的呢,首先获取CompletableFuture中的stack对象,是个null值,然后将封装阻塞等待线程的Signaller对象中的next变量指向这个stack,
可以看到这里直接进行了赋值,没有走cas操作,因为这个Signaller对象,不同的线程是不一样的,相互独立的,所以没有必要进行cas操作,但是下一步UNSAFE.compareAndSwapObject(this,STACK, h, c),将CompletableFuture中的stack变量替换为封装自身线程的Signaller对象时,会有竞争存在,所以需要走cas保证线程安全性,执行失败的线程会重新走一遍while循环走到这里,直到设置成功。下面一个if是判断线程中断的,我们跳过,最后一个if会执行managedBlock方法
里面核心是 !blocker.isReleasable() && !blocker.block()这个条件,isReleasable方法恒为false,我们看block方法,
可以看到里面将线程挂起了。当方法执行完成后会将他们唤醒,回到get方法里,当拿到结果后执行reportGet方法
前面说过,如果方法执行结果为null或者没有返回结果时,会返回一个封装了null值的AltResult对象,这里判断对象是否是AltResult对象,如果是,则判断封装的结果是否为null,是则返回null,不是则抛出相应的异常,如果都不是返回正常执行的结果。
所以,如果有多个线程阻塞等待的话,所有线程会先排个队,用stack数据结构保存起来,然后各自将自己挂起,等方法运行结束后会将这些阻塞线程全部唤醒。
以上即为CompletableFuture方法运行和获取值的原理,还有很多其他的方法,但是底层方法基本都是上面说的那几个。