Java8 lamda

lamda可以认为是匿名函数，用过scala的都很熟悉

 

Java8 supplier

supplier，

Suppliers represent a function that accepts no arguments and produce a result of some arbitrary type.

 

A Future represents the pending result of an asynchronous computation. It offers a method — get — that returns the result of the computation when it's done.

The problem is that a call to get is blocking until the computation is done. This is quite restrictive and can quickly make the asynchronous computation pointless.

传统的Future，提供异步计算的可能性

不过Future的结果需要用get去获取，这个过程是block的或者轮询，所以这样又限制了异步

 

Beside implementing the Future interface, CompletableFuture also implements the CompletionStage interface.

A CompletionStage is a promise. It promises that the computation eventually will be done.

The great thing about the CompletionStage is that it offers a vast selection of methods that let you attach callbacks that will be executed on completion.

This way we can build systems in a non-blocking fashion.

所以我们需要CompletableFuture，除了future接口，还实现CompletionStage接口

CompletionStage的作用是，提供给你很多方法，让你可以attach各种callbacks用于future执行完成后的followup

这样我们更容易build non-blocking的系统

例子，

Let's start with the absolute basics — creating a simple asynchronous computation.

CompletableFuture.supplyAsync(this::sendMsg);

It's as easy as that.

supplyAsync takes a Supplier containing the code we want to execute asynchronously — in our case the sendMsgmethod.

异步执行supplier的逻辑，什么是supplier见上面

If you've worked a bit with Futures in the past, you may wonder where the Executor went. If you want to, you can still define it as a second argument. However, if you leave it out it will be submitted to the ForkJoinPool.commonPool().

ForkJoinPool.commonPool

Java 8为ForkJoinPool添加了一个通用线程池，这个线程池用来处理那些没有被显式提交到任何线程池的任务。它是ForkJoinPool类型上的一个静态元素，它拥有的默认线程数量等于运行计算机上的处理器数量

commonPool可以理解为由JVM管理的一个通用线程池，谁都可以用

 

加上callback

In the first example, we simply sent a message asynchronously by executing sendMsg in its own thread.

Now let's add a callback where we notify about how the sending of the message went.

CompletableFuture.supplyAsync(this::sendMsg)                   .thenAccept(this::notify);

thenAccept is one of many ways to add a callback. It takes a Consumer — in our case notify — which handles the result of the preceding computation when it's done.

thenAccept是加callback的一种方式，当然有很多种加callback的方式；

这里thenAccept是一种Consumer function，consumer即接受参数没有返回的一种function

在这里notify会作为consumer会接收sendMsg是输出作为输入来继续处理

加上多个callbacks

If you want to continue passing values from one callback to another, thenAccept won't cut it since Consumer doesn't return anything.

To keep passing values, you can simply use thenApply instead.

上面说了thenAccept是一种Consumer，是不会有返回值的，所以如果要级联多个callback，需要用thenApply，它是既有输入参数，又有返回值

thenApply takes a Function which accepts a value, but also return one.

To see how this works, let's extend our previous example by first finding a receiver.

CompletableFuture.supplyAsync(this::findReceiver)                     .thenApply(this::sendMsg)                     .thenAccept(this::notify);

Now the asynchronous task will first find a receiver, then send a message to the receiver before it passes the result on to the last callback to notify.

thenCompose

Until now, sendMsg has been a normal blocking function. Let's now assume that we got a sendMsgAsync method that returns aCompletionStage.

If we kept using thenApply to compose the example above, we would end up with nested CompletionStages.

CompletableFuture.supplyAsync(this::findReceiver)                   .thenApply(this::sendMsgAsync);// Returns type CompletionStage
        
         
          >
         
        

这个例子会出现多个异步的嵌套，

所以这里用thenCompose，去除掉嵌套

CompletableFuture.supplyAsync(this::findReceiver)                   .thenCompose(this::sendMsgAsync);// Returns type CompletionStage
        

异步的执行callback

Until now all our callbacks have been executed on the same thread as their predecessor.

If you want to, you can submit the callback to the ForkJoinPool.commonPool() independently instead of using the same thread as the predecessor. This is done by using the async suffix version of the methods CompletionStage offers.

我们也可以异步的执行callback，

CompletableFuture
        
          receiver              = CompletableFuture.supplyAsync(this::findReceiver);receiver.thenApply(this::sendMsg);  receiver.thenApply(this::sendOtherMsg);
        

这个例子中，两次的消息发送都是在当前线程中完成的，我们也可以给加上async前缀，让他们异步的执行，

CompletableFuture
        
          receiver              = CompletableFuture.supplyAsync(this::findReceiver);receiver.thenApplyAsync(this::sendMsg);  receiver.thenApplyAsync(this::sendMsg);
        

 

异常处理

CompletableFuture.supplyAsync(this::failingMsg)                   .exceptionally(ex -> new Result(Status.FAILED))                 .thenAccept(this::notify);

exceptionally gives us a chance to recover by taking an alternative function that will be executed if preceding calculation fails with an exception.

This way succeeding callbacks can continue with the alternative result as input.

If you need more flexibility, check out whenComplete and handle for more ways of handling errors.

 

thenCombine

thenCombine allows us to register a BiFunction callback depending on the result of two CompletionStages.

CompletableFuture
        
          to =      CompletableFuture.supplyAsync(this::findReceiver);CompletableFuture
         
           text =      CompletableFuture.supplyAsync(this::createContent);to.thenCombine(text, this::sendMsg);
         
        

上面的是both的case

下面的是either的case

Let’s say you have two sources of finding a receiver. You’ll ask both, but will be happy with the first one returning with a result.

CompletableFuture
        
          firstSource =      CompletableFuture.supplyAsync(this::findByFirstSource);CompletableFuture
         
           secondSource =      CompletableFuture.supplyAsync(this::findBySecondSource);firstSource.acceptEither(secondSource, this::sendMsg);