SpringBoot JAVA

Java8——异步编程

Java8——异步编程

Posted by 不学无数 on August 13, 2019

Java8——异步编程

异步编程

所谓异步其实就是实现一个无需等待被调用函数的返回值而让操作继续运行的方法

创建任务并执行任务

无参创建

 CompletableFuture<String> noArgsFuture = new CompletableFuture<>();

传入相应任务,无返回值

runAsync方法可以在后台执行异步计算，但是此时并没有返回值。持有一个Runnable对象。

CompletableFuture noReturn = CompletableFuture.runAsync(()->{
    //执行逻辑,无返回值
});

传入相应任务,有返回值

此时我们看到返回的是CompletableFuture<T>此处的T就是你想要的返回值的类型。其中的Supplier<T>是一个简单的函数式接口。

CompletableFuture<String> hasReturn = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        return "hasReturn";
    }
});

此时可以使用lambda表达式使上面的逻辑更加清晰

CompletableFuture<String> hasReturnLambda = CompletableFuture.supplyAsync(TestFuture::get);

private static String get() {
    return "hasReturnLambda";
}

获取返回值

异步任务也是有返回值的，当我们想要用到异步任务的返回值时，我们可以调用CompletableFuture 的get()阻塞，直到有异步任务执行完有返回值才往下执行。

我们将上面的get()方法改造一下，使其停顿十秒时间。

private static String get() {
    System.out.println("Begin Invoke getFuntureHasReturnLambda");
    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {

    }
    System.out.println("End Invoke getFuntureHasReturnLambda");
    return "hasReturnLambda";
}

然后进行调用

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<String> funtureHasReturnLambda = (CompletableFuture<String>) getFuntureHasReturnLambda();
    System.out.println("Main Method Is Invoking");
    funtureHasReturnLambda.get();
    System.out.println("Main Method End");
}

可以看到输出如下，只有调用get()方法的时候才会阻塞当前线程。

Main Method Is Invoking
Begin Invoke getFuntureHasReturnLambda
End Invoke getFuntureHasReturnLambda
Main Method End

自定义返回值

除了等待异步任务返回值以外，我们也可以在任意时候调用complete()方法来自定义返回值。


CompletableFuture<String> funtureHasReturnLambda = (CompletableFuture<String>) getFuntureHasReturnLambda();
System.out.println("Main Method Is Invoking");
new Thread(()->{
    System.out.println("Thread Is Invoking ");
    try {
        Thread.sleep(1000);
        funtureHasReturnLambda.complete("custome value");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("Thread End ");
}).run();
String value = funtureHasReturnLambda.get();
System.out.println("Main Method End value is "+ value);

我们可以发现输出是新起线程的输出值，当然这是因为我们的异步方法设置了等待10秒，如果此时异步方法等待1秒，新起的线程等待10秒，那么输出的值就是异步方法中的值了。

Main Method Is Invoking
Begin Invoke getFuntureHasReturnLambda
Thread Is Invoking 
Thread End 
Main Method End value is custome value

按顺序执行异步任务

如果有一个异步任务的完成需要依赖前一个异步任务的完成，那么该如何写呢？是调用get()方法获得返回值以后然后再执行吗？这样写有些麻烦，CompletableFuture 为我们提供了方法来完成我们想要顺序执行一些异步任务的需求。thenApply 、thenAccept 、thenRun 这三个方法。这三个方法的区别就是。

方法名 是否可获得前一个任务的返回值 是否有返回值
thenApply 能获得 有
thenAccept 能获得 无
thenRun 不可获得 无

所以一般来说thenAccept 、thenRun 这两个方法在调用链的最末端使用。接下来我们用真实的例子感受一下。


//thenApply  可获取到前一个任务的返回值,也有返回值
CompletableFuture<String> seqFutureOne = CompletableFuture.supplyAsync(()-> "seqFutureOne");
CompletableFuture<String> seqFutureTwo = seqFutureOne.thenApply(name -> name + " seqFutureTwo");
System.out.println(seqFutureTwo.get());


//thenAccept  可获取到前一个任务的返回值,但是无返回值
CompletableFuture<Void> thenAccept = seqFutureOne
        .thenAccept(name -> System.out.println(name + "thenAccept"));
System.out.println("-------------");
System.out.println(thenAccept.get());

//thenRun 获取不到前一个任务的返回值,也无返回值
System.out.println("-------------");
CompletableFuture<Void> thenRun = seqFutureOne.thenRun(() -> {
    System.out.println("thenRun");
});
System.out.println(thenRun.get());

返回的信息如下

seqFutureOne seqFutureTwo
seqFutureOnethenAccept
-------------
null
-------------
thenRun
null

thenApply和thenApplyAsync的区别

我们可以发现这三个方法都带有一个后缀为Async的方法，例如thenApplyAsync。那么带Async的方法和不带此后缀的方法有什么不同呢？我们就以thenApply和thenApplyAsync 两个方法进行对比，其他的和这个一样的。

这两个方法区别就在于谁去执行这个任务，如果使用thenApplyAsync ，那么执行的线程是从ForkJoinPool.commonPool()中获取不同的线程进行执行，如果使用thenApply ，如果supplyAsync 方法执行速度特别快，那么thenApply 任务就是主线程进行执行，如果执行特别慢的话就是和supplyAsync 执行线程一样。接下来我们通过例子来看一下，使用sleep方法来反应supplyAsync 执行速度的快慢。

//thenApply和thenApplyAsync的区别
System.out.println("-------------");
CompletableFuture<String> supplyAsyncWithSleep = CompletableFuture.supplyAsync(()->{
    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "supplyAsyncWithSleep Thread Id : " + Thread.currentThread();
});
CompletableFuture<String> thenApply = supplyAsyncWithSleep
        .thenApply(name -> name + "------thenApply Thread Id : " + Thread.currentThread());
CompletableFuture<String> thenApplyAsync = supplyAsyncWithSleep
        .thenApplyAsync(name -> name + "------thenApplyAsync Thread Id : " + Thread.currentThread());
System.out.println("Main Thread Id: "+ Thread.currentThread());
System.out.println(thenApply.get());
System.out.println(thenApplyAsync.get());
System.out.println("-------------No Sleep");
CompletableFuture<String> supplyAsyncNoSleep = CompletableFuture.supplyAsync(()->{
    return "supplyAsyncNoSleep Thread Id : " + Thread.currentThread();
});
CompletableFuture<String> thenApplyNoSleep = supplyAsyncNoSleep
        .thenApply(name -> name + "------thenApply Thread Id : " + Thread.currentThread());
CompletableFuture<String> thenApplyAsyncNoSleep = supplyAsyncNoSleep
        .thenApplyAsync(name -> name + "------thenApplyAsync Thread Id : " + Thread.currentThread());
System.out.println("Main Thread Id: "+ Thread.currentThread());
System.out.println(thenApplyNoSleep.get());
System.out.println(thenApplyAsyncNoSleep.get());

我们可以看到输出为

-------------
Main Thread Id: Thread[main,5,main]
supplyAsyncWithSleep Thread Id : Thread[ForkJoinPool.commonPool-worker-1,5,main]------thenApply Thread Id : Thread[ForkJoinPool.commonPool-worker-1,5,main]
supplyAsyncWithSleep Thread Id : Thread[ForkJoinPool.commonPool-worker-1,5,main]------thenApplyAsync Thread Id : Thread[ForkJoinPool.commonPool-worker-1,5,main]
-------------No Sleep
Main Thread Id: Thread[main,5,main]
supplyAsyncNoSleep Thread Id : Thread[ForkJoinPool.commonPool-worker-2,5,main]------thenApply Thread Id : Thread[main,5,main]
supplyAsyncNoSleep Thread Id : Thread[ForkJoinPool.commonPool-worker-2,5,main]------thenApplyAsync Thread Id : Thread[ForkJoinPool.commonPool-worker-2,5,main]

可以看到supplyAsync方法执行速度慢的话thenApply方法执行线程和supplyAsync 执行线程相同，如果supplyAsync 方法执行速度快的话，那么thenApply方法执行线程和Main方法执行线程相同。

组合CompletableFuture

将两个CompletableFuture组合到一起有两个方法

  1. thenCompose()：当第一个任务完成时才会执行第二个操作
  2. thenCombine()：两个异步任务全部完成时才会执行某些操作

thenCompose() 用法

我们定义两个异步任务，假设第二个定时任务需要用到第一个定时任务的返回值。

public static CompletableFuture<String> getTastOne(){
    return CompletableFuture.supplyAsync(()-> "topOne");
}

public static CompletableFuture<String> getTastTwo(String s){
    return CompletableFuture.supplyAsync(()-> s + "  topTwo");
}

我们利用thenCompose()方法进行编写

CompletableFuture<String> thenComposeComplet = getTastOne().thenCompose(s -> getTastTwo(s));
System.out.println(thenComposeComplet.get());

输出就是

topOne  topTwo

如果还记得前面的thenApply()方法的话，应该会想这个利用thenApply()方法也是能够实现类似的功能的。

//thenApply
CompletableFuture<CompletableFuture<String>> thenApply = getTastOne()
        .thenApply(s -> getTastTwo(s));
System.out.println(thenApply.get().get());

但是我们发现返回值是嵌套返回的一个类型，而想要获得最终的返回值需要调用两次get()

thenCombine() 用法

例如我们此时需要计算两个异步方法返回值的和。求和这个操作是必须是两个异步方法得出来值的情况下才能进行计算，因此我们可以用thenCombine()方法进行计算。

CompletableFuture<Integer> thenComposeOne = CompletableFuture.supplyAsync(() -> 192);
CompletableFuture<Integer> thenComposeTwo = CompletableFuture.supplyAsync(() -> 196);
CompletableFuture<Integer> thenComposeCount = thenComposeOne
        .thenCombine(thenComposeTwo, (s, y) -> s + y);
System.out.println(thenComposeCount.get());

此时thenComposeOne 和thenComposeTwo 都完成时才会调用传给thenCombine 方法的回调函数。

组合多个CompletableFuture

在上面我们用thenCompose()和thenCombine()两个方法将两个CompletableFuture 组装起来，如果我们想要将任意数量的CompletableFuture 组合起来呢?可以使用下面两个方法进行组合。

  • allOf()：等待所有CompletableFuture 完后以后才会运行回调函数
  • anyOf()：只要其中一个CompletableFuture 完成，那么就会执行回调函数。注意此时其他的任务也就不执行了。

接下来演示一下两个方法的用法

//allOf()
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> 1);
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture<Integer> three = CompletableFuture.supplyAsync(() -> 3);
CompletableFuture<Integer> four = CompletableFuture.supplyAsync(() -> 4);
CompletableFuture<Integer> five = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> six = CompletableFuture.supplyAsync(() -> 6);

CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(one, two, three, four, five, six);
voidCompletableFuture.thenApply(v->{
    return Stream.of(one,two,three,four, five, six)
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
}).thenAccept(System.out::println);

CompletableFuture<Void> voidCompletableFuture1 = CompletableFuture.runAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (Exception e) {

    }
    System.out.println("1");
});

我们定义了6个CompletableFuture 等待所有的CompletableFuture等待所有任务完成以后然后将其值输出。

anyOf()的用法

CompletableFuture<Void> voidCompletableFuture1 = CompletableFuture.runAsync(() -> {
try {
    Thread.sleep(1000);
} catch (Exception e) {

}
System.out.println("voidCompletableFuture1");
});

CompletableFuture<Void> voidCompletableFutur2 = CompletableFuture.runAsync(() -> {
try {
    Thread.sleep(2000);
} catch (Exception e) {

}
System.out.println("voidCompletableFutur2");
});

CompletableFuture<Void> voidCompletableFuture3 = CompletableFuture.runAsync(() -> {
try {
    Thread.sleep(3000);
} catch (Exception e) {

}
System.out.println("voidCompletableFuture3");
});

CompletableFuture<Object> objectCompletableFuture = CompletableFuture
    .anyOf(voidCompletableFuture1, voidCompletableFutur2, voidCompletableFuture3);
objectCompletableFuture.get();

这里我们定义了3个CompletableFuture进行一些耗时的任务，此时第一个CompletableFuture会率先完成。打印结果如下。

voidCompletableFuture1

异常处理

我们了解了CompletableFuture 如何异步执行，如何组合不同的CompletableFuture ，如何顺序执行CompletableFuture 。那么接下来还有一个重要的一步，就是在执行异步任务时发生异常的话该怎么办。我们先写个例子。


CompletableFuture.supplyAsync(()->{
    //发生异常
    int i = 10/0;
    return "Success";
}).thenRun(()-> System.out.println("thenRun"))
.thenAccept(v -> System.out.println("thenAccept"));

CompletableFuture.runAsync(()-> System.out.println("CompletableFuture.runAsync"));

执行结果为，我们发现只要执行链中有一个发生了异常，那么接下来的链条也就不执行了，但是主流程下的其他CompletableFuture还是会运行的。

CompletableFuture.runAsync

exceptionally()

我们可以使用exceptionally进行异常的处理

//处理异常

CompletableFuture<String> exceptionally = CompletableFuture.supplyAsync(() -> {
    //发生异常
    int i = 10 / 0;
    return "Success";
}).exceptionally(e -> {
    System.out.println(e);
    return "Exception has Handl";
});
System.out.println(exceptionally.get());

打印如下，可以发现其接收值是异常信息，也能够返回自定义返回值。

java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
Exception has Handl

handle()

调用handle()方法也能够捕捉到异常并且自定义返回值，他和exceptionally()方法不同一点是handle()方法无论发没发生异常都会被调用。例子如下

System.out.println("-------有异常-------");
CompletableFuture.supplyAsync(()->{
    //发生异常
    int i = 10/0;
    return "Success";
}).handle((response,e)->{
    System.out.println("Exception:" + e);
    System.out.println("Response:" + response);
    return response;
});

System.out.println("-------无异常-------");
CompletableFuture.supplyAsync(()->{
    return "Sucess";
}).handle((response,e)->{
    System.out.println("Exception:" + e);
    System.out.println("Response:" + response);
    return response;
});

打印如下，我们可以看到在没有发生异常的时候handle()方法也被调用了

-------有异常-------
Exception:java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
Response:null
-------无异常-------
Exception:null
Response:Sucess

源代码地址

参考文章

CATALOG
    FEATURED TAGS
    SpringBoot Java 设计模式 MyBatis 遇到的坑 瞎扯淡 Mybatis 源码解析 Spring 技巧 容器 JAVA Redis 数据库 Tomcat NIO Netty 事务 网络 并发 JVM 架构
    FRIENDS