文章原创,转载请注明出处:http://abc08010051.iteye.com/blog/2409693
后面会再修改一下,让文章读起来更好读,现在的版本还比较粗糙
CompletableFuture是java 1.8提供的一个新类,是对Future的增强,吸收了guava异步线程的特点,可以实现一系列的异步线程操作,很多常规的用法网上有很多博客,这里说说部分代码的实现:
这是CompletableFuture的基本结构
CompletableFuture基本属性方法
volatile Object result; // Either the result or boxed AltResult volatile Completion stack; // Top of Treiber stack of dependent actions
result用来存放线程返回的结果
stack 行为上就是一个栈的功能,先进后出,用来存放要执行的动作,这个在单个异步线程返回时是没用的,多个线程等待的时候才排上用场
Completion基本属性方法
volatile Completion next; // Treiber stack link abstract CompletableFuture<?> tryFire(int mode);
next链式结构,存放下一个
tryFire 方法,主要返回一个依赖的Completion
名词解释:
dependent:依赖,多个线程操作时,如何等待每个线程都完成才返回,主要是依靠这个依赖,没处理完就会调用依赖的postComplete()方法向上传递
source:源,用户自定义的线程的CompletableFuture
1 单一使用
/** * Returns a new CompletableFuture that is asynchronously completed * by a task running in the {@link ForkJoinPool#commonPool()} with * the value obtained by calling the given Supplier. * * @param supplier a function returning the value to be used * to complete the returned CompletableFuture * @param <U> the function's return type * @return the new CompletableFuture */ public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { return asyncSupplyStage(asyncPool, supplier); } 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; } static final class AsyncSupply<T> extends ForkJoinTask<Void> implements Runnable, AsynchronousCompletionTask { CompletableFuture<T> dep; Supplier<T> fn; AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) { this.dep = dep; this.fn = fn; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} public final boolean exec() { run(); return true; } public void run() { //CompletableFuture句柄,把Supplier的返回值放到CompletableFuture的result属性中,当前线程的执行是在默认的线程池中执行,在外部可以获取 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(); } } } /** * Pops and tries to trigger all reachable dependents. Call only * when known to be done. */ 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; //循环遍历CompletableFuture的stack属性,Completion是一个链式的操作,如果有下一个,触发下一个Completion的tryFire方法 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; } } }
等待获取结果:
public T get() throws InterruptedException, ExecutionException { Object r; return reportGet((r = result) == null ? waitingGet(true) : r); } private Object waitingGet(boolean interruptible) { Signaller q = null; boolean queued = false; int spins = -1; Object r; //循环获取result属性,判断是否为空,不为空获取到结果,跳出while循环 while ((r = result) == null) { if (spins < 0) //多个线程在允许,就给spins赋值256,然后循环递减,如果此时还没有返回值,则走下面的else分支 spins = (Runtime.getRuntime().availableProcessors() > 1) ? 1 << 8 : 0; // Use brief spin-wait on multiprocessors else if (spins > 0) { if (ThreadLocalRandom.nextSecondarySeed() >= 0) --spins; } else if (q == null) //创建等待信号线程 q = new Signaller(interruptible, 0L, 0L); else if (!queued) //替换stack属性,把替换是否成功的结果赋值给queued queued = tryPushStack(q); else if (interruptible && q.interruptControl < 0) {//允许中断,并且q.interruptControl = 1,不会走此分支, 下面的循环出现出现线程中断会走此分支 q.thread = null; cleanStack(); return null; } else if (q.thread != null && result == null) {//如果结果没有返回,会进入当前分支 try { //循环判断q是否释放,等待一直到满足Signaller释放条件(主要判断是否超时),上面Signaller的构造方法中,deadline为0, 不会因为超时释放,只有线程中断的时候才会释放 ForkJoinPool.managedBlock(q); } catch (InterruptedException ie) {//如果发生线程中断,把Signaller的interruptControl置为-1,等到下一个循环使用 q.interruptControl = -1; } } } if (q != null) {//信号线程不为null, 如果Signaller的中断控制标记位小于0,则返回null或者线程中断 q.thread = null; if (q.interruptControl < 0) { if (interruptible) r = null; // report interruption else Thread.currentThread().interrupt(); } } //传递给下一个Completion,没有则不执行 postComplete(); return r; } private static <T> T reportGet(Object r) throws InterruptedException, ExecutionException { //根据不同的情况做返回值的包装 if (r == null) // by convention below, null means interrupted throw new InterruptedException(); if (r instanceof AltResult) { Throwable x, cause; if ((x = ((AltResult)r).ex) == null) return null; if (x instanceof CancellationException) throw (CancellationException)x; if ((x instanceof CompletionException) && (cause = x.getCause()) != null) x = cause; throw new ExecutionException(x); } @SuppressWarnings("unchecked") T t = (T) r; return t; }
2 等待多个线程执行完成再做返回
//demo , stageRunnable是一个实现Runnable类型的变量 CompletableFuture future = CompletableFuture.allOf(CompletableFuture.runAsync(stageRunnable), CompletableFuture.runAsync(stageRunnable), CompletableFuture.runAsync(stageRunnable)); System.out.println(JSON.toJSONString(future)); future.get(); public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { return andTree(cfs, 0, cfs.length - 1); } //此方法是一个递归方法, 二分法把两个任务执行一个等待,每次二分都会创建一个CompletableFuture的depency static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs, int lo, int hi) { CompletableFuture<Void> d = new CompletableFuture<Void>(); if (lo > hi) // empty d.result = NIL; else { CompletableFuture<?> a, b; int mid = (lo + hi) >>> 1; if ((a = (lo == mid ? cfs[lo] : andTree(cfs, lo, mid))) == null || (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : andTree(cfs, mid+1, hi))) == null) throw new NullPointerException(); if (!d.biRelay(a, b)) {//a,b两个子任务没有全部完成,走此分支 BiRelay<?,?> c = new BiRelay<>(d, a, b);//创建一个Completion,一个依赖,两个source a.bipush(b, c);//把c推送到a,b的stack属性当中去 c.tryFire(SYNC);//BiRelay触发实际操作 } } return d; } //根据方法名直译的意思:是否两个传播都已经完成;两个任务有任何一个未完成,则返回false, 只有全部完成的时候才会返回true boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) { Object r, s; Throwable x; if (a == null || (r = a.result) == null || b == null || (s = b.result) == null) return false; if (result == null) { if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) completeThrowable(x, r); else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null) completeThrowable(x, s); else completeNull(); } return true; }
相关推荐
Lua 源码赏析.pdf, 其中有着非常棒的lua源代码资源可供学习。下载文件为该书的百度云连接地址。
103协议源码赏析
Lua中文教程+源码赏析,两小时学会Lua编程,pdf版本,清晰度高。
Larbin搜索引擎源码赏析[整理].pdf
Larbin搜索引擎源码赏析[借鉴].pdf
future源码Java中具有CompletableFuture的异步编程 介绍 CompletableFuture API是用于Java异步编程的高级API。 该API支持将多个异步计算流水线化(也称为链接或合并)成单个结果,而不会造成嵌套回调(“ callback ...
skynet的源码赏析
从源码层面,剖析挖掘互联网行业主流技术的底层实现原理,为广大开发者 “提升技术深度” 提供便利。目前开放 Spring 全家桶,Mybatis、Netty、Dubbo 框架,及 Redis、Tomcat 中间件等,让我们一起开拓新的领地,...
coobjc是为iOS平台打造的开源协程开发框架,支持Objective-C和Swift,同时提供了cokit库为Foundation和UIKit中的部分API提供了协程化支持,本文将为大家详细介绍coobjc的设计理念及核心优势。
SpringMVC URL 与 Controller 方法初始化源码流程 Visio 文档 文档可以直接通过Visio进行编辑,方便二次修改、学习
sm2-3-4代码赏析,有不会的可以问我,我看了一下这份代码,觉得学习还是不错的,4的可以拿去直接用,2的需要改一下
Spark-2.3.1源码解读。 Spark Core源码阅读 Spark Context 阅读要点 Spark的缓存,变量,shuffle数据等清理及...PIDController源码赏析及 back pressure 实现思路 Streaming Context重点摘要 checkpoint 必知必会
programming in lua 4th,lua程序设计第4版,我学第一版的时候是200多页,现在2016年出的第四版还是200多页,更新了一些内容。还有国内大神云峰编写的《lua源码鉴赏》,分享出来一块学习进步吧。
skynet源代码详细解析,让你快速了解SKYNET的全貌。
最近有一点空闲,想续写我那本 Lua 源码欣赏。按我心里的计划,还有大约 6 章。虚拟机、字节码持久化、C API 、解释器、GC、库函数。 新添了一章关于虚拟机的,所以重新读了一遍相关源码。发现 Lua 5.2 比上一版...
云风写的,讲了一些实现的细节,适合想要深入的朋友,分就不要了,没分的也可以下
YY多开源码YY多开器源码源码YY多开器源码源码YY多开器源码源码
订餐网,外卖网源码,带积分商城,商家系统,外卖网站建设! 系统特点: 周密策划、项目为先 "项目指导技术,技术服从项目",这是我们一贯秉承的原则,也是我们与其他系统开发商、网站建设公司的本质区别所在!我们...
源码01 销售管理系统 源码02 彩票分析系统 源码03 餐饮管理系统 源码04 C#点名程序 源码05 象棋游戏 源码06 变色球游戏 源码07 多功能计算器 源码08 记事本 源码09 简易画图程序 源码10 成绩管理系统 源码11 BBS论坛...