1.概述CompletableFuture是jdk1.8引入的实现类 。扩展了Future和CompletionStage,是一个可以在任务完成阶段触发一些操作Future 。简单的来讲就是可以实现异步回调 。
2.为什么引入CompletableFuture对于jdk1.5的Future,虽然提供了异步处理任务的能力,但是获取结果的方式很不优雅,还是需要通过阻塞(或者轮训)的方式 。如何避免阻塞呢?其实就是注册回调 。
业界结合观察者模式实现异步回调 。也就是当任务执行完成后去通知观察者 。比如Netty的ChannelFuture,可以通过注册监听实现异步结果的处理 。
Netty的ChannelFuturepublic Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {checkNotNull(listener, "listener");synchronized (this) {addListener0(listener);}if (isDone()) {notifyListeners();}return this;}private boolean setValue0(Object objResult) {if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {if (checkNotifyWaiters()) {notifyListeners();}return true;}return false;}通过addListener方法注册监听 。如果任务完成,会调用notifyListeners通知 。
CompletableFuture通过扩展Future,引入函数式编程,通过回调的方式去处理结果 。
3.功能CompletableFuture的功能主要体现在他的CompletionStage 。
可以实现如下等功能
- 转换(thenCompose)
- 组合(thenCombine)
- 消费(thenAccept)
- 运行(thenRun) 。
- 带返回的消费(thenApply)
消费使用执行结果 。运行则只是运行特定任务 。具体其他功能大家可以根据需求自行查看 。
CompletableFuture借助CompletionStage的方法可以实现链式调用 。并且可以选择同步或者异步两种方式 。
这里举个简单的例子来体验一下他的功能 。
public static void thenApply() {ExecutorService executorService = Executors.newFixedThreadPool(2);CompletableFuture cf = CompletableFuture.supplyAsync(() -> {try {//Thread.sleep(2000);} catch (Exception e) {e.printStackTrace();}System.out.println("supplyAsync " + Thread.currentThread().getName());return "hello";}, executorService).thenApplyAsync(s -> {System.out.println(s + "world");return "hhh";}, executorService);cf.thenRunAsync(() -> {System.out.println("ddddd");});cf.thenRun(() -> {System.out.println("ddddsd");});cf.thenRun(() -> {System.out.println(Thread.currentThread());System.out.println("dddaewdd");});}执行结果supplyAsync pool-1-thread-1helloworlddddddddddsdThread[main,5,main]dddaewdd根据结果我们可以看到会有序执行对应任务 。注意:
如果是同步执行cf.thenRun 。他的执行线程可能main线程,也可能是执行源任务的线程 。如果执行源任务的线程在main调用之前执行完了任务 。那么cf.thenRun方法会由main线程调用 。
这里说明一下,如果是同一任务的依赖任务有多个:
- 如果这些依赖任务都是同步执行 。那么假如这些任务被当前调用线程(main)执行,则是有序执行,假如被执行源任务的线程执行,那么会是倒序执行 。因为内部任务数据结构为LIFO 。
- 如果这些依赖任务都是异步执行,那么他会通过异步线程池去执行任务 。不能保证任务的执行顺序 。
4.源码追踪创建CompletableFuture创建的方法有很多,甚至可以直接new一个 。我们来看一下supplyAsync异步创建的方法 。
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) {return asyncSupplyStage(screenExecutor(executor), supplier);}static Executor screenExecutor(Executor e) {if (!useCommonPool && e == ForkJoinPool.commonPool())return asyncPool;if (e == null) throw new NullPointerException();return e;}入参Supplier,带返回值的函数 。如果是异步方法,并且传递了执行器,那么会使用传入的执行器去执行任务 。否则采用公共的ForkJoin并行线程池,如果不支持并行,新建一个线程去执行 。这里我们需要注意ForkJoin是通过守护线程去执行任务的 。所以必须有非守护线程的存在才行 。
asyncSupplyStage方法
static <U> CompletableFuture<U> asyncSupplyStage(Executor e,Supplier<U> f) {if (f == null) throw new NullPointerException();CompletableFuture<U> d = new CompletableFuture<U>();e.execute(new AsyncSupply<U>(d, f));return d;}这里会创建一个用于返回的CompletableFuture 。然后构造一个AsyncSupply,并将创建的CompletableFuture作为构造参数传入 。
那么,任务的执行完全依赖AsyncSupply 。
AsyncSupply#run
public void run() {CompletableFuture<T> d; Supplier<T> f;if ((d = dep) != null && (f = fn) != null) {dep = null; fn = null;if (d.result == null) {try {d.completeValue(f.get());} catch (Throwable ex) {d.completeThrowable(ex);}}d.postComplete();}}- 该方法会调用Supplier的get方法 。并将结果设置到CompletableFuture中 。我们应该清楚这些操作都是在异步线程中调用的 。
d.postComplete方法就是通知任务执行完成 。触发后续依赖任务的执行,也就是实现CompletionStage的关键点 。
thenAcceptAsync方法
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {return uniAcceptStage(asyncPool, action);}private CompletableFuture<Void> uniAcceptStage(Executor e,Consumer<? super T> f) {if (f == null) throw new NullPointerException();CompletableFuture<Void> d = new CompletableFuture<Void>();if (e != null || !d.uniAccept(this, f, null)) {# 1UniAccept<T> c = new UniAccept<T>(e, d, this, f);push(c);c.tryFire(SYNC);}return d;}上面提到过 。thenAcceptAsync是用来消费CompletableFuture的 。该方法调用uniAcceptStage 。uniAcceptStage逻辑:
- 构造一个CompletableFuture,主要是为了链式调用 。
- 如果为异步任务,直接返回 。因为源任务结束后会触发异步线程执行对应逻辑 。
- 如果为同步任务(e==null),会调用d.uniAccept方法 。这个方法在这里逻辑:如果源任务完成,调用f,返回true 。否则进入if代码块(Mark 1) 。
- 如果是异步任务直接进入if(Mark 1) 。
- 构造一个UniAccept,将其push入栈 。这里通过CAS实现乐观锁实现 。
- 调用c.tryFire方法 。
final CompletableFuture<Void> tryFire(int mode) {CompletableFuture<Void> d; CompletableFuture<T> a;if ((d = dep) == null ||!d.uniAccept(a = src, fn, mode > 0 ? null : this))return null;dep = null; src = https://tazarkount.com/read/null; fn = null;return d.postFire(a, mode);}- 会调用d.uniAccept方法 。其实该方法判断源任务是否完成,如果完成则执行依赖任务,否则返回false 。
- 如果依赖任务已经执行,调用d.postFire,主要就是Fire的后续处理 。根据不同模式逻辑不同 。
这里强调一下d.uniAccept方法的第三个参数 。
如果是异步调用(mode>0),传入null 。否则传入this 。
区别看下面代码 。c不为null会调用c.claim方法 。
try {if (c != null && !c.claim())return false;@SuppressWarnings("unchecked") S s = (S) r;f.accept(s);completeNull();} catch (Throwable ex) {completeThrowable(ex);}final boolean claim() {Executor e = executor;if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {if (e == null)return true;executor = null; // disablee.execute(this);}return false;}claim方法是逻辑:- 如果异步线程为null 。说明同步,那么直接返回true 。最后上层函数会调用f.accept(s)同步执行任务 。
- 如果异步线程不为null,那么使用异步线程去执行this 。
public final void run(){tryFire(ASYNC);}看完上面的逻辑,我们基本理解依赖任务的逻辑 。其实就是先判断源任务是否完成,如果完成,直接在对应线程执行以来任务(如果是同步,则在当前线程处理,否则在异步线程处理)
如果任务没有完成,直接返回,因为等任务完成之后会通过postComplete去触发调用依赖任务 。
postComplete方法
final void postComplete() {/** On each step, variable f holds current dependents to pop* and run.It is extended along only one path at a time,* pushing others to avoid unbounded recursion.*/CompletableFuture<?> f = this; Completion h;while ((h = f.stack) != null ||(f != this && (h = (f = this).stack) != null)) {CompletableFuture<?> d; Completion t;if (f.casStack(h, t = h.next)) {if (t != null) {if (f != this) {pushStack(h);continue;}h.next = null;// detach}f = (d = h.tryFire(NESTED)) == null ? this : d;}}}在源任务完成之后会调用 。其实逻辑很简单,就是迭代堆栈的依赖任务 。调用h.tryFire方法 。NESTED就是为了避免递归死循环 。因为FirePost会调用postComplete 。如果是NESTED,则不调用 。
堆栈的内容其实就是在依赖任务创建的时候加入进去的 。上面我们已经提到过 。
4.总结基本上述源码已经分析了逻辑 。
因为涉及异步等操作,我们需要理一下(这里针对全异步任务):
- 创建CompletableFuture成功之后会通过异步线程去执行对应任务 。
- 如果CompletableFuture还有依赖任务(异步),会将任务加入到CompletableFuture的堆栈保存起来 。以供后续完成后执行依赖任务 。
主要是考虑代码的复用 。所以逻辑相对难理解 。
postComplete方法会被源任务线程执行完源任务后调用 。同样也可能被依赖任务线程后调用 。
执行依赖任务的方法主要就是靠tryFire方法 。因为这个方法可能会被多种不同类型线程触发,所以逻辑也绕一点 。(其他依赖任务线程、源任务线程、当前依赖任务线程)
- 如果是当前依赖任务线程,那么会执行依赖任务,并且会通知其他依赖任务 。
- 如果是源任务线程,和其他依赖任务线程,则将任务转换给依赖线程去执行 。不需要通知其他依赖任务,避免死递归 。
来源:blog.csdn.net/weixin_39332800/article/
- 春季老年人吃什么养肝?土豆、米饭换着吃
- 三八妇女节节日祝福分享 三八妇女节节日语录
- 老人谨慎!选好你的“第三只脚”
- 校方进行了深刻的反思 青岛一大学生坠亡校方整改校规
- 脸皮厚的人长寿!有这特征的老人最长寿
- 长寿秘诀:记住这10大妙招 100%增寿
- 春季老年人心血管病高发 3条保命要诀
- 眼睛花不花要看四十八 老年人怎样延缓老花眼
- 香槟然能防治老年痴呆症? 一天三杯它人到90不痴呆
- 老人手抖的原因 为什么老人手会抖
