completablefuture是jdk1.8引入的实现类。扩展了future和completionstage,是一个可以在任务完成阶段触发一些操作future。简单的来讲就是可以实现异步回调。
对于jdk1.5的future,虽然提供了异步处理任务的能力,但是获取结果的方式很不优雅,还是需要通过阻塞(或者轮训)的方式。如何避免阻塞呢?其实就是注册回调。
业界结合观察者模式实现异步回调。也就是当任务执行完成后去通知观察者。比如netty的channelfuture,可以通过注册监听实现异步结果的处理。
netty的channelfuture
public promi<v> addlistener(genericfuturelistener<? extends future<? super v>> listener) { checknotnull(listener, "listener"); synchronized (this) { addlistener0(listener); } if (isdone()) { notifylisteners(); } return this;}private boolean tvalue0(object objresult) { if (result_updater.compareandt(this, null, objresult) || result_updater.compareandt(this, uncancellable, objresult)) { if (checknotifywaiters()) { notifylisteners(); } return true; } return fal;}
通过addlistener方法注册监听。如果任务完成,会调用notifylis喉结核teners通知。
completablefuture通过扩展future,引入函数式编程,通过回调的方式去处理结果。
completablefuture的功能主要体现在他的completionstage。
可以实现如下等功能
转换(thencompo)组合(thencombine)消费(thenaccept)运行(thenrun)。带返回的消费(thenapply)消费和运行的区别:
消费使用执行结果。运行则只是运行特定任务。具体其他功能大家可以根据需求自行查看。
completablefuture借助completionstage的方法可以实现链式调用。并且可以选择同步或者异步两种方式。
这里举个简单的例子来体验一下他的功能。
public static void thenapply() { executorrvice executorrvice = 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"; }, executorrvice).thenapplyasync(s -> { system.out.println(s + "world"); return "hhh"; }, executorrvice); 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-1
helloworld
ddddd
ddddsd
thread[main,5,main]
dddaewdd
根据结果我们可以看到会有序执行对应任务。
注意:
如果是同步执行cf.thenrun。他的执行线程可能main线程,也可能是执行源任务的线程。如果执行源任务的线程在main调用之前执行完了任务。那么cf.thenrun方法会由main线程调用。
这里说明一下,如果是同一任务的依赖任务有多个:
如果这些依赖任务都是同步执行。那么假如这些任务被当前调用线程(main)执行,则是有序执行,假如被执行源任务的线程执行,那么会是倒序执行。因为内部任务数据结构为lifo。如果这些依赖任务都是异步执行,那么他会通过异步线程池去执行任务。不能保证任务的执行顺序。上面的结论是通过阅读源代码得到的。下面我们深入源代码。
创建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 (!ucommonpool && e == forkjoinpool.commonpool()) return asyncpool; if (e == null) throw new nullpointerexception(); return e;}
入参supplier,带返回值的函数。如果是异步方法,并且传递了执行器,那么会使用传入的执行器去执行任务。否则采用公共的forkjoin并行线程池,如果不支持并行,新建一个线程去执行。
这里我们需要注意forkjoin是通过守护线程去执行任务的。所以必须有非守护线程的存在才行。
asy创维显示器怎么样ncsupplystage方法
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(); }}
1.该方法会调用supplier的get方法。并将结果设置到completablefuture中。我们应该清楚这些操作都是在异步线程中调用的。
2.d.postcomplete方法就是通知任务执行完成。触发后续依赖任务的执行,也就是实现completionstage的关键点。
在看postcomplete方法之前我们先来看一下创建依赖任务的逻辑。
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)) { # 2021最新一期青年大学生答案1 uniaccept<t> c = new uniaccept<t>(e, d, this, f); push(c); c.tryfire(sync); } return d;}
上面提到过。thenacceptasync是用来消费completablefuture的。该方法调用uniacceptstage。
uniacceptstage逻辑:
1.构造一个completablefuture,主要是为了链式调用。
2.如果为异步任务,直接返回。因为源任务结束后会触发异步线程执行对应逻辑。
3.如果为同步任务(e==null),会调用d.uniaccept方法。这个方法在这里逻辑:如果源任务完成,调用f,返回true。否则进入if代码块(mark 1)。
4.如果是异步任务直接进入if(mark 1)。
mark1逻辑:
1.构造一个uniaccept,将其push入栈。这里通过cas实现乐观锁实现。
2.调用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 = null; fn = null; return d.postfire(a, mode);}
1.会调用d.uniaccept方法。其实该方法判断源任务是否完成,如果完成则执行依赖任务,否则返回fal。
2.如果依赖任务已经执行,调用d.postfire,主要就是fire的后续处理。根据不同模式逻辑不同。
这里简单说一下,其实mode有同步异步,和迭代。迭代为了避免无限递归。
这里强调一下d.uniaccept方法的第三个参数。
如果是异步调用(mode>0),传入null。否则传入this。
区别看下面代码。c不为null会调用c.claim方法。
try { if (c != null && !c.claim()) return fal; @suppresswarnings("unchecked") s s = (s) r; f.accept(s); completenull();} catch (throwable ex) { completethrowable(ex);}final boolean claim() { executor e = executor; if (compareandtforkjointasktag((short)0, (short)1)) { if (e == null) return true; executor = null; // disable e.execute(this); } return fal;}
claim方法是逻辑:
如果异步线程为null。说明同步,那么直接返回true。最后上层函数会调用f.accept(s)同步执行任务。如果异步线程不为null,那么使用异步线程去执行this。this的run任务如下。也就是在异步线程同步调用tryfire方法。达到其被异步线程执行的目的。
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,则不调用。
堆栈的内容其实就是在依赖任务创建的时候加入进去的。上面我们已经提到过。
基本上述源码已经分析了逻辑。
因为涉及异步等操作,我们需要理一下(这里针对全异步任务):
1.创建completablefuture成功之后会通过异步线程去执行对应任务。
2.如果completablefuture还有依赖任务(异步),会将任务加入到completablefuture的堆栈保存起来。以供后续完成后执行依赖任务。
当然,创建依赖任务并不只是将其加入堆栈。如果源任务在创建依赖任务的时候已经执行完成,那么当前线程会触发依赖任务的异步线程直接处理依赖任务。并且会告诉堆栈其他的依赖任务源任务已经完成。
主要是考虑代码的复用。所以逻辑相对难理解。
postcomplete方法会被源任务线程执行完源任务后调用。同样也可能被依赖任务线程后调用。
执行依赖任务的方法主要就是靠tryfire方法。因为这个方法可能会被多种不同类型线程触发,所以逻辑也绕一点。(其他依赖任务线程、源任务线程、当前依赖任务线程)
如果是当前依赖任务线程,那么会执行依赖任务,并且会通知其他依赖任务。如果是时事新闻源任务线程,和其他依赖任务线程,则将任务转换给依赖线程去执行。不需要通知其他依赖任务,避免死递归。不得不说doug lea的编码,真的是艺术。代码的复用性全体现在逻辑上了。
到此这篇关于java8中completablefuture使用场景与实现原理的文章就介绍到这了,更多相关completablefuture使用场景与原理内容请搜索www.887551.com以前的文章或继续浏览下面的相关文章希望大家以后多多支持www.887551.com!
本文发布于:2023-04-04 22:05:57,感谢您对本站的认可!
本文链接:https://www.wtabcd.cn/fanwen/zuowen/42b02c63e69fbf12be48ee6de4c11b6e.html
版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,我们将在24小时内删除。
本文word下载地址:Java8中CompletableFuture使用场景与实现原理.doc
本文 PDF 下载地址:Java8中CompletableFuture使用场景与实现原理.pdf
留言与评论(共有 0 条评论) |