CompletableFutures使用

非阻塞异步编程

所谓非阻塞异步编程方法，简单地说，就是不用等待返回结果的多线程的回调方法的封装。相比于阻塞式异步回调方法，非阻塞异步编程方法使用一个监听器，这样在使用回调的过程中，就能够自动感知到调用结果，从而实现了高并发的调用方法。

使用阻塞式调用，会有大量的时间耗费在等待之中。Java 开发语 Java1.8 之后 ，提供了非阻塞的回调处理方法，它封装在CompletableFuture 类中。CompletableFuture 提供了非常强大的程序扩展能力可以帮助我们简化异步编程的复杂性，并通过非阻塞的回调方式处理调用结果。 CompletableFuture 不但提供了非阻塞的异步调用的方法，还可以将处理结果进行转换和结合使用。

场景

比如有这样的业务场景：

在用户信息查询接口中需要返回：用户名称、性别、等级、头像、积分、成长值等信息。

而用户名称、性别、等级、头像在用户服务中，积分在积分服务中，成长值在成长值服务中。为了汇总这些数据统一返回，需要另外提供一个对外接口服务。

于是，用户信息查询接口需要调用用户查询接口、积分查询接口 和 成长值查询接口，然后汇总数据统一返回。

调用过程如下图所示：

调用远程接口总耗时 530ms = 200ms + 150ms + 180ms

显然这种串行调用远程接口性能是非常不好的，调用远程接口总的耗时为所有的远程接口耗时之和。

那么如何优化远程接口性能呢？

上面说到，既然串行调用多个远程接口性能很差，为什么不改成并行呢？

如下图所示：

调用远程接口总耗时 200ms = 200ms（即耗时最长的那次远程接口调用）

API 方法分类

创建类

  • completeFuture 可以用于创建默认返回值
  • runAsync 异步执行，无返回值
  • supplyAsync 异步执行，有返回值
  • anyOf 任意一个执行完成，就可以进行下一步动作
  • allOf 全部完成所有任务，才可以进行下一步任务

状态取值类

  • join 合并结果，等待
  • get 合并等待结果，可以增加超时时间；get 和 join 区别，join 只会抛出 unchecked 异常，get 会返回具体的异常
  • getNow 如果结果计算完成或者异常了，则返回结果或异常；否则，返回 valueIfAbsent 的值
  • isCancelled
  • isCompletedExceptionally
  • isDone

控制类 用于主动控制 CompletableFuture 的完成行为

  • complete
  • completeExceptionally
  • cancel

接续类 用于注入回调行为

  • thenApply, thenApplyAsync
  • thenAccept, thenAcceptAsync
  • thenRun, thenRunAsync
  • thenCombine, thenCombineAsync
  • thenAcceptBoth, thenAcceptBothAsync
  • runAfterBoth, runAfterBothAsync
  • applyToEither, applyToEitherAsync
  • acceptEither, acceptEitherAsync
  • runAfterEither, runAfterEitherAsync
  • thenCompose, thenComposeAsync
  • whenComplete, whenCompleteAsync
  • handle, handleAsync
  • exceptionally

记忆规则

  1. 以 Async 结尾的方法，都是异步方法，对应的没有 Async 则是同步方法，一般都是一个异步方法对应一个同步方法。

  2. 以 Async 后缀结尾的方法，都有两个重载的方法，一个是使用内容的 forkjoin 线程池，一种是使用自定义线程池

  3. 以 run 开头的方法，其入口参数一定是无参的，并且没有返回值，类似于执行 Runnable 方法。

  4. 以 supply 开头的方法，入口也是没有参数的，但是有返回值

  5. 以 Accept 开头或者结尾的方法，入口参数是有参数，但是没有返回值

  6. 以 Apply 开头或者结尾的方法，入口有参数，有返回值

  7. 带有 either 后缀的方法，表示谁先完成就消费谁

API使用

创建 CompletableFuture

创建 CompletableFuture，其实就是将我们要煮的米饭，委托给电饭煲；要煮米饭，我们要准备这么几件事情，其一我们要制定制作米饭的方式，其二，我们要指定电饭煲。除此之外，我们也可以委托其他的事情，最后可以通过 all 或者 any 进行组合。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 异步任务，无返回值，采用内部的forkjoin线程池
CompletableFuture c1 = CompletableFuture
.runAsync(()->{System.out.println("打开开关，开始制作，就不用管了")});

// 异步任务，无返回值，使用自定义的线程池
CompletableFuture c11 = CompletableFuture
.runAsync(()->{System.out.println("打开开关，开始制作，就不用管了")},newSingleThreadExecutor());

// 异步任务，有返回值，使用内部默认的线程池
CompletableFuture<String> c2 = CompletableFuture
.supplyAsync(()->{System.out.println("清洗米饭");return "干净的米饭";});

// 只要有一个完成，则完成，有一个抛出异常，则携带异常
CompletableFuture.anyOf(c1,c2);

// 必须等待所有的future全部完成才可以
CompletableFuture.allOf(c1,c2);

取值与状态

1
2
3
4
5
6
7
// 常用的是下面的这几种
// 不抛出异常，阻塞的等待
future.join()
// 有异常则抛出异常，阻塞的等待，无限等待
future.get()
// 有异常则抛出异常，最长等待1个小时，一个小时之后，如果还没有数据，则异常。
future.get(1,TimeUnit.Hours)

控制 CompletableFuture 执行

1
2
3
4
5
6
7
// 3种方式：
// 完成
future.complete("米饭");
// 异常
future.completeExceptionally();
// 取消，参数并没有实际意义，没任何卵用。
future.cancel(false);

接续 CompletableFuture + (Runnable,Consumer,Function)

1
2
3
4
5
6
7
8
9
CompletableFuture future = CompletableFuture.supplyAsync(()->{
    System.out.println("投放和清洗制作米饭的材料");
    return "干净的没有新冠病毒的大米";
}).thenAcceptAsync(result->{
    System.out.println("通电，设定模式，开始煮米饭");
}).thenRunAsync(()->{
    System.out.println("米饭做好了，可以吃了");
})

接续 CompletableFuture + CompletableFuture

假如蒸米饭和、热牛奶、炒菜等已经是 3 个不同的 CompletableFuture，可以使用接续方式，将两个或者多个 CompletableFuture 组合在一起使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
CompletableFuture rice = CompletableFuture.supplyAsync(()->{
    System.out.println("开始制作米饭，并获得煮熟的米饭");
    return "煮熟的米饭";
})

// 煮米饭的同时呢，我又做了牛奶
CompletableFuture mike = CompletableFuture.supplyAsync(()->{
    System.out.println("开始热牛奶，并获得加热的牛奶");
    return "加热的牛奶";
});

// 我想两个都好了，才吃早饭，thenCombineAsync有入参，有返回值
mike.thenCombineAsync(rice,(m,r)->{
    System.out.println("我收获了早饭："+m+","+r);
    return m+r;
})
// 有入参，无返回值
mike.thenAcceptBothAsync(rice,(m,r)->{
   System.out.println("我收获了早饭："+m+","+r); 
});
// 无入参
mike.runAfterBothAsync(rice,()->{
   System.out.println("我收获了早饭"); 
});

// 或者直接连接两个CompletableFuture
rice.thenComposeAsync(r->CompletableFuture.supplyAsync(()->{
    System.out.println("开始煮牛奶");
    System.out.println("同时开始煮米饭");
    return "mike";
}))

接续 CompletableFuture + 处理结果

如果我们只想做结果处理，也没有其他的接续动作，并且我们想要判断异常的情况，那么可以用这种接续方式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
// whenCompleteAsync：处理完成或异常，无返回值
// handleAsync：处理完成或异常，有返回值

CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).whenCompleteAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
})
// 有返回值
CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).handleAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
    return "准备冷一冷再吃米饭";
})

// 异常处理
CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).handleAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
    return "准备冷一冷再吃米饭";
}).exceptionally((exception)->{
    // 前置动作必须的是一个有返回值的操作
    return "";
});

SpringBoot使用

线程池配置类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
 
//线程池配置类
@Configuration
public class TaskPoolConfig {
 
    @Bean("myTaskExecutor")
    public Executor myTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);//核心线程数量，线程池创建时候初始化的线程数
        executor.setMaxPoolSize(15);//最大线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setQueueCapacity(200);//缓冲队列，用来缓冲执行任务的队列
        executor.setKeepAliveSeconds(60);//当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
        executor.setThreadNamePrefix("myTask-");//设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);//用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
        executor.setAwaitTerminationSeconds(60);//该方法用来设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住。
        //线程池对拒绝任务的处理策略：这里采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在 execute 方法的调用线程中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

Service层

1
2
3
4
5
6
7
8
9
10
11
12
@Service
public class UserService {
 
    public String getUserName(){
        return "biandan";
    }
 
    public String getSign(){
        return "让天下没有难写的代码";
    }
 
}

案例 1：使用 get 获取返回结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@RestController
public class UserController {
 
    @Autowired
    private UserService userService;
 
    @Autowired
    @Qualifier("myTaskExecutor")
    private Executor executor;
 
    //案例1：使用 supplyAsync 异步获取返回结果
    @GetMapping("/getUserInfo")
    public String getUserInfo() throws ExecutionException, InterruptedException {
        CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程名：" + Thread.currentThread().getName());
            return userService.getUserName();
        }, executor);
        String result = supplyAsync.get();
        return result;
    }
 
}

案例 2：使用 thenApply 方法将结果返回给其它函数调用

1
2
3
4
5
6
7
8
9
10
11
//案例2：使用 thenApply 方法将结果返回给其它函数调用
@GetMapping("/getUserInfo2")
public String getUserInfo2() throws ExecutionException, InterruptedException {
    CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
        System.out.println("getUserName 当前线程名：" + Thread.currentThread().getName());
        return userService.getUserName();
    }, executor);
    CompletableFuture<String> thenApply = supplyAsync.thenApply(r -> r + " 提供给函数");
    String result = thenApply.get();
    return result;
}

说明：使用上面的对象，调用 thenApply 即可将返回结果给其它函数调用。

①如果其它函数需要获取结果，但该函数并不需要返回值，可以定义成 Void，并使用的是 thenAccept 方法如下：

1
CompletableFuture<Void> thenAccept = supplyAsync.thenAccept(s->{});

②如果其它函数既不需要获取结果，也不需要返回值，只想单纯的在前面的方法执行完成后，再执行一段逻辑，可以使用 thenRun。如下：

1
CompletableFuture<Void> thenRun = supplyAsync.thenRun(()->{});

案例 3：使用 thenCompose 链接两个调用

1
2
3
4
5
6
7
8
9
10
11
12
13
//案例3：使用 thenCompose 链接两个调用
@GetMapping("/getUserInfo3")
public String getUserInfo3() throws ExecutionException, InterruptedException {
    CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
        System.out.println("getUserName 当前线程名：" + Thread.currentThread().getName());
        return userService.getUserName();
    }, executor)
            .thenCompose(result -> CompletableFuture.supplyAsync(() -> {
                System.out.println("getSign 当前线程名：" + Thread.currentThread().getName());
                return result + " 说 " + userService.getSign();
            }, executor));
    return supplyAsync.get();
}

测试结果：biandan 说 让天下没有难写的代码

说明：使用 thenCompose 可以将上一个的返回结果给下一个函数使用。

如果要链接可完成的 CompletableFuture 方法，那么最好使用 thenCompose ()

案例 4：使用 thenCombine 将两个结果结合起来，给另一个函数调用

1
2
3
4
5
6
7
8
9
//案例4：使用 thenCombine 将两个结果结合起来，给另一个调用
@GetMapping("/getUserInfo4")
public String getUserInfo4() throws ExecutionException, InterruptedException {
    CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> {
        return "biandan";
    }).thenCombine(CompletableFuture.supplyAsync(() -> " 说 "), (r1, r2) -> r1 + r2)
            .thenCompose(p -> CompletableFuture.supplyAsync(() -> p + "让天下没有难写的代码"));
    return result.get();
}

案例 5：allOf 多种结合的调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//案例5：allof 多种结合的调用
@GetMapping("/getUserInfo5")
public String getUserInfo5() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1
            = CompletableFuture.supplyAsync(() -> "biandan");
    CompletableFuture<String> future2
            = CompletableFuture.supplyAsync(() -> "说");
    CompletableFuture<String> future3
            = CompletableFuture.supplyAsync(() -> "让天下没有难写的代码");
 
    //返回的是 void 类型
    CompletableFuture<Void> combinedFuture_1
            = CompletableFuture.allOf(future1, future2, future3);
 
    //使用 join 方法手动获取结果
    String combinedFuture_2 = Stream.of(future1, future2, future3)
            .map(CompletableFuture::join)
            .collect(Collectors.joining(" "));
 
    return combinedFuture_2;
}

注意 CompletableFuture.allOf 的返回类型是 CompletableFuture。这种方法的局限性在于它不能返回所有 Supplier 的组合结果。我们必须从未来手动获取结果。使用 CompletableFuture.join () 方法获取。

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

徘徊在原子与字节的边缘<br><br>将更好的自己呈现给世界<br><br><br><br>修心至要，事上磨练<br><br>一蓑衣，一壶酒，一曲长歌，一剑天涯<br><br>见自己，见天地，见众生<br><br>