从JDK 8开始,在Concurrent包中提供了一个强大的异步编程工具CompletableFuture。在JDK8之前,异步编程可以通过线程池和Future来实现,但功能还不够强大。
CompletableFuture实现了Future接口,所以它也具有Future的特性:调用get()方法会阻塞在那,直到结果返回。
另外1个线程调用complete方法完成该Future,则所有阻塞在get()方法的线程都将获得返回结果。
上面的例子是一个空的任务,下面尝试提交一个真的任务,然后等待结果返回。
例1:runAsync(Runnable)
CompletableFuture.runAsync(...)传入的是一个Runnable接口。
例2: supplyAsync(Supplier)
例2和例1的区别在于,例2的任务有返回值。没有返回值的任务,提交的是Runnable,返回的是CompletableFuture;有返回值的任务,提交的是 Supplier,返回的是CompletableFuture。Supplier和前面的Callable很相似。
通过上面两个例子可以看出,在基本的用法上,CompletableFuture和Future很相似,都可以提交两类任务:一类是无返回值的,另一类是有返回值的。
对于 Future,在提交任务之后,只能调用 get()等结果返回;但对于 CompletableFuture,可以在结果上面再加一个callback,当得到结果之后,再接着执行callback。
例1:thenRun(Runnable)
该案例最后不能获取到结果,只会得到一个null。
例2:thenAccept(Consumer)
上述代码在thenAccept中可以获取任务的执行结果,接着进行处理。
例3:thenApply(Function)
三个例子都是在任务执行完成之后,接着执行回调,只是回调的形式不同:
而参数接收的是前一个任务,即 supplyAsync(...)这个任务的返回值。因此这里只能用supplyAsync,不能用runAsync。因为runAsync没有返回值,不能为下一个链式方法传入参数。
例1:thenCompose
在上面的例子中,thenApply接收的是一个Function,但是这个Function的返回值是一个通常的基本数据类型或一个对象,而不是另外一CompletableFuture。如果 Function 的返回值也是一个CompletableFuture,就会出现嵌套的CompletableFuture。考虑下面的例子:
如果希望返回值是一个非嵌套的CompletableFuture,可以使用thenCompose:
下面是thenCompose方法的接口定义:
CompletableFuture中的实现:
从该方法的定义可以看出,它传入的参数是一个Function类型,并且Function的返回值必须是CompletionStage的子类,也就是CompletableFuture类型。
例2:thenCombine
thenCombine方法的接口定义如下,从传入的参数可以看出,它不同于thenCompose。
第1个参数是一个CompletableFuture类型,第2个参数是一个方法,并且是一个BiFunction,也就是该方法有2个输入参数,1个返回值。
从该接口的定义可以大致推测,它是要在2个 CompletableFuture 完成之后,把2个CompletableFuture的返回值传进去,再额外做一些事情。实例如下:
上面的thenCompose和thenCombine只能组合2个CompletableFuture,而接下来的allOf 和anyOf 可以组合任意多个CompletableFuture。方法接口定义如下所示。
首先,这两个方法都是静态方法,参数是变长的CompletableFuture的集合。其次,allOf和anyOf的区别,前者是“与”,后者是“或”。
allOf的返回值是CompletableFuture类型,这是因为每个传入的CompletableFuture的返回值都可能不同,所以组合的结果是无法用某种类型来表示的,索性返回Void类型。
anyOf 的含义是只要有任意一个 CompletableFuture 结束,就可以做接下来的事情,而无须像AllOf那样,等待所有的CompletableFuture结束。
但由于每个CompletableFuture的返回值类型都可能不同,任意一个,意味着无法判断是什么类型,所以anyOf的返回值是CompletableFuture类型。
通过上面的例子可以总结出,提交给CompletableFuture执行的任务有四种类型:Runnable、Consumer、Supplier、Function。下面是这四种任务原型的对比。
runAsync 与 supplierAsync 是 CompletableFuture 的静态方法;而 thenAccept、thenAsync、thenApply是CompletableFutre的成员方法。
因为初始的时候没有CompletableFuture对象,也没有参数可传,所以提交的只能是Runnable或者Supplier,只能是静态方法;
通过静态方法生成CompletableFuture对象之后,便可以链式地提交其他任务了,这个时候就可以提交Runnable、Consumer、Function,且都是成员方法。
CompletableFuture不仅实现了Future接口,还实现了CompletableStage接口。
CompletionStage接口定义的正是前面的各种链式方法、组合方法,如下所示。
关于CompletionStage接口,有几个关键点要说明:
其他方法,诸如thenCompose、thenCombine也是类似的原理。
CompletableFuture中任务的执行依靠ForkJoinPool:
通过上面的代码可以看到,asyncPool是一个static类型,supplierAsync、asyncSupplyStage也都是static方法。Static方法会返回一个CompletableFuture类型对象,之后就可以链式调用,CompletionStage里面的各个方法。
ForkJoinPool接受的任务是ForkJoinTask 类型,而我们向CompletableFuture提交的任务是Runnable/Supplier/Consumer/Function 。因此,肯定需要一个适配机制,把这四种类型的任务转换成ForkJoinTask,然后提交给ForkJoinPool,如下图所示:
为了完成这种转换,在CompletableFuture内部定义了一系列的内部类,下图是CompletableFuture的各种内部类的继承体系。
在 supplyAsync(...)方法内部,会把一个 Supplier 转换成一个 AsyncSupply,然后提交给ForkJoinPool执行;
在runAsync(...)方法内部,会把一个Runnable转换成一个AsyncRun,然后提交给ForkJoinPool执行;
在 thenRun/thenAccept/thenApply 内部,会分别把 Runnable/Consumer/Function 转换成UniRun/UniAccept/UniApply对象,然后提交给ForkJoinPool执行;
除此之外,还有两种 CompletableFuture 组合的情况,分为“与”和“或”,所以有对应的Bi和Or类型的Completion类型。
下面的代码分别为 UniRun、UniApply、UniAccept 的定义,可以看到,其内部分别封装了Runnable、Function、Consumer。
下面以CompletableFuture.supplyAsync(...).thenApply(...).thenRun(...)链式代码为例,分析整个执行过程。
第1步:CompletableFuture future1=CompletableFuture.supplyAsync(...)
在上面的代码中,关键是构造了一个AsyncSupply对象,该对象有三个关键点:
ForkJoinPool执行一个ForkJoinTask类型的任务,即AsyncSupply。该任务的输入就是Supply,输出结果存放在CompletableFuture中。
第2步:CompletableFuture future2=future1.thenApply(...)
第1步的返回值,也就是上面代码中的 CompletableFuture d,紧接着调用其成员方法thenApply:
我们知道,必须等第1步的任务执行完毕,第2步的任务才可以执行。因此,这里提交的任务不可能立即执行,在此处构建了一个UniApply对象,也就是一个ForkJoinTask类型的任务,这个任务放入了第1个任务的栈当中。
每一个CompletableFuture对象内部都有一个栈,存储着是后续依赖它的任务,如下面代码所示。这个栈也就是Treiber Stack,这里的stack存储的就是栈顶指针。
上面的UniApply对象类似于第1步里面的AsyncSupply,它的构造方法传入了4个参数:
UniApply对象被放入了第1步的CompletableFuture的栈中,在第1步的任务执行完成之后,就会从栈中弹出并执行。如下代码:
ForkJoinPool执行上面的AsyncSupply对象的run()方法,实质就是执行Supplier的get()方法。执行结果被塞入了 CompletableFuture d 当中,也就是赋值给了 CompletableFuture 内部的Object result变量。
调用d.postComplete(),也正是在这个方法里面,把第2步压入的UniApply对象弹出来执行,代码如下所示。
第3步:CompletableFuture future3=future2.thenRun()
第3步和第2步的过程类似,构建了一个 UniRun 对象,这个对象被压入第2步的CompletableFuture所在的栈中。第2步的任务,当执行完成时,从自己的栈中弹出UniRun对象并执行。
综上所述:
通过supplyAsync/thenApply/thenRun,分别提交了3个任务,每1个任务都有1个返回值对象,也就是1个CompletableFuture。这3个任务通过2个CompletableFuture完成串联。后1个任务,被放入了前1个任务的CompletableFuture里面,前1个任务在执行完成时,会从自己的栈中,弹出下1个任务执行。如此向后传递,完成任务的链式执行。
在上面的代码中,我们分析了thenApply,还有一个与之对应的方法是thenApplyAsync。这两个方法调用的是同一个方法,只不过传入的参数不同。
对于上一个任务已经得出结果的情况:
如果e != null表示是thenApplyAsync,需要调用ForkJoinPool的execute方法,该方法:
通过上面的代码可以看到:
同理,thenRun与thenRunAsync、thenAccept与thenAcceptAsync的区别与此类似。
如果任务只是链式执行,便不需要在每个CompletableFuture里面设1个栈了,用1个指针使所有任务组成链表即可。
但实际上,任务不只是链式执行,而是网状执行,组成 1 张图。如下图所示,所有任务组成一个有向无环图:
任务一执行完成之后,任务二、任务三可以并行,在代码层面可以写为:future1.thenApply(任务二),future1.thenApply(任务三);
任务四在任务二执行完成时可开始执行;
任务五要等待任务二、任务三都执行完成,才能开始,这里是AND关系;
任务六在任务三执行完成时可以开始执行;
对于任务七,只要任务四、任务五、任务六中任意一个任务结束,就可以开始执行。
总而言之,任务之间是多对多的关系:1个任务有_n_个依赖它的后继任务;1个任务也有_n_个它依赖的前驱任务。
这样一个有向无环图,用什么样的数据结构表达呢?AND和OR的关系又如何表达呢?
有几个关键点:
任务五的确会被触发二次,但它会判断任务二、任务三的结果是不是都完成,如果只完成其中一个,它就不会执行。
任何一个多元操作,都能被转换为多个二元操作的叠加。如上图所示,假如任务一AND任务二AND任务三 ==> 任务四,那么它可以被转换为右边的形式。新建了一个AND任务,这个AND任务和任务三再作为参数,构造任务四。OR的关系,与此类似。
此时,thenCombine的内部实现原理也就可以解释了。thenCombine用于任务一、任务二执行完成,再执行任务三。
下面以allOf方法为例,看一下有向无环计算图的内部运作过程:
上面的方法是一个递归方法,输入是一个CompletableFuture对象的列表,输出是一个具有AND关系的复合CompletableFuture对象。
最关键的代码如上面加注释部分所示,因为d要等a,b都执行完成之后才能执行,因此d会被分别压入a,b所在的栈中。
下图为allOf内部的运作过程。假设allof的参数传入了future1、future2、future3、future4,则对应四个原始任务。
生成BiRelay1、BiRelay2任务,分别压入future1/future2、future3/future4的栈中。无论future1或future2完成,都会触发BiRelay1;无论future3或future4完成,都会触发BiRelay2;
生成BiRelay3任务,压入future5/future6的栈中,无论future5或future6完成,都会触发BiRelay3任务。
BiRelay只是一个中转任务,它本身没有任务代码,只是参照输入的两个future是否完成。如果完成,就从自己的栈中弹出依赖它的BiRelay任务,然后执行。
原网址: 访问
创建于: 2022-10-28 00:25:04
目录: default
标签: 无
未标明原创文章均为采集,版权归作者所有,转载无需和我联系,请注明原出处,南摩阿彌陀佛,知识,不只知道,要得到
最新评论