👊 【Spring 技术原理】异步编程机制以及功能分析讲解

李浩宇/Alex

关注

发布于: 5 小时前

本文内容

Future 模式介绍以及核心思想
核心线程数、最大线程数的区别，队列容量代表什么；
ThreadPoolTaskExecutor 饱和策略；
SpringBoot 异步编程实战，搞懂代码的执行逻辑。

Future 模式

异步编程在处理耗时操作以及多任务处理的场景下非常有用，我们可以更好的让我们的系统利用好机器的 CPU 和内存，提高它们的利用率。
多线程设计模式有很多种，Future 模式是多线程开发中较为常见的一种设计模式，本文也是基于这种模式来说明 SpringBoot 对于异步编程的知识。

实战之前我先简单介绍一下 Future 模式的核心思想吧！。

Future 的核心思想

Future 模式的核心思想是异步调用。当我们执行一个方法时，假如这个方法中有多个耗时的任务需要同时去做，而且又不着急等待这个结果时可以让客户端立即返回然后，后台慢慢去计算任务。当然你也可以选择等这些任务都执行完了，再返回给客户端。

SpringBoot 异步编程实战

如果我们需要在 Spring/SpringBoot 实现异步编程的话，通过 Spring 提供的两个注解会让这件事情变的非常简单。

@EnableAsync：通过在配置类或者 Main 类上加 @EnableAsync 开启对异步方法的支持。
@Async 可以作用在类上或者方法上，作用在类上代表这个类的所有方法都是异步方法。

TaskExecutor

很多人对于 TaskExecutor 不是太了解，所以我们花一点篇幅先介绍一下这个东西。从名字就能看出它是任务的执行者，它领导执行着线程来处理任务，就像司令官一样，而我们的线程就好比一只只军队一样，这些军队可以异步对敌人进行打击。
Spring 提供了 TaskExecutor 接口作为任务执行者的抽象,它和 java.util.concurrent 包下的 Executor 接口很像。稍微不同的 TaskExecutor 接口用到了 Java 8 的语法 @FunctionalInterface 声明这个接口是一个函数式接口。

org.springframework.core.task.TaskExecutor@FunctionalInterfacepublic interface TaskExecutor extends Executor {    void execute(Runnable var1);}

复制代码

如果没有自定义 Executor, Spring 将创建一个 SimpleAsyncTaskExecutor 并使用它。

自定义 AsyncConfigurer

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.annotation.AsyncConfigurer;import org.springframework.scheduling.annotation.EnableAsync;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Executor;

@Configuration@EnableAsyncpublic class AsyncConfig implements AsyncConfigurer {
  private static final int CORE_POOL_SIZE = 6;  private static final int MAX_POOL_SIZE = 10;  private static final int QUEUE_CAPACITY = 100;
  @Bean  public Executor taskExecutor() {    // Spring 默认配置是核心线程数大小为1，最大线程容量大小不受限制，队列容量也不受限制。    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();    // 核心线程数    executor.setCorePoolSize(CORE_POOL_SIZE);    // 最大线程数    executor.setMaxPoolSize(MAX_POOL_SIZE);    // 队列大小    executor.setQueueCapacity(QUEUE_CAPACITY);    // 当最大池已满时，此策略保证不会丢失任务请求，但是可能会影响应用程序整体性能。    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());    executor.setThreadNamePrefix("My ThreadPoolTaskExecutor-");    executor.initialize();    return executor;  }}

复制代码

ThreadPoolTaskExecutor 常见概念

Core Pool Size : 核心线程数线程数定义了最小可以同时运行的线程数量。
Queue Capacity : 当新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，信任就会被存放在队列中。
Maximum Pool Size : 当队列中存放的任务达到队列容量的时候，当前可以同时运行的线程数量变为最大线程数。

一般情况下不会将队列大小设为：Integer.MAX_VALUE,也不会将核心线程数和最大线程数设为同样的大小，这样的话最大线程数的设置都没什么意义了，你也无法确定当前 CPU 和内存利用率具体情况如何。

如果队列已满并且当前同时运行的线程数达到最大线程数的时候，如果再有新任务过来会发生什么呢？Spring 默认使用的是 ThreadPoolExecutor.AbortPolicy（ThreadPoolExecutor 将抛出 RejectedExecutionException 来拒绝新来的任务，这代表你将丢失对这个任务的处理。）

对于可伸缩的应用程序，建议使用 ThreadPoolExecutor.CallerRunsPolicy，当最大池被填满时，此策略为我们提供可伸缩队列。

ThreadPoolTaskExecutor 饱和策略定义:

如果当前同时运行的线程数量达到最大线程数量时，ThreadPoolTaskExecutor 定义一些策略:

ThreadPoolExecutor.AbortPolicy：抛出 RejectedExecutionException 来拒绝新任务的处理。
ThreadPoolExecutor.CallerRunsPolicy：调用执行自己的线程运行任务。您不会任务请求。但是这种策略会降低对于新任务提交速度，影响程序的整体性能。
另外，这个策略喜欢增加队列容量。如果您的应用程序可以承受此延迟并且你不能任务丢弃任何一个任务请求的话，你可以选择这个策略。
ThreadPoolExecutor.DiscardPolicy：不处理新任务，直接丢弃掉。
ThreadPoolExecutor.DiscardOldestPolicy：此策略将丢弃最早的未处理的任务请求。

编写一个异步的方法

给这个方法加上了 @Async 注解来告诉 Spring 它是一个异步的方法。另外，这个方法的返回值 CompletableFuture.completedFuture(results)这代表我们需要返回结果，也就是说程序必须把任务执行完成之后再返回给用户。

请留意 completableFutureTask 方法中的第一行打印日志这句代码，后面分析程序中会用到，很重要！

import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.scheduling.annotation.Async;import org.springframework.stereotype.Service;import java.util.ArrayList;import java.util.Arrays;import java.util.List;import java.util.concurrent.CompletableFuture;import java.util.stream.Collectors;
@Servicepublic class AsyncService {  private static final Logger logger = LoggerFactory.getLogger(AsyncService.class);  private List<String> movies =      new ArrayList<>(          Arrays.asList(              "Forrest Gump",              "Titanic",              "Spirited Away",              "The Shawshank Redemption",              "Zootopia",              "Farewell ",              "Joker",              "Crawl"));  /** 示范使用：找到特定字符/字符串开头的电影 */  @Async  public CompletableFuture<List<String>> completableFutureTask(String start) {    // 打印日志    logger.warn(Thread.currentThread().getName() + "start this task!");    // 找到特定字符/字符串开头的电影    List<String> results =        movies.stream().filter(movie -> movie.startsWith(start)).collect(Collectors.toList());    // 模拟这是一个耗时的任务    try {      Thread.sleep(1000L);    } catch (InterruptedException e) {      e.printStackTrace();    }    //返回一个已经用给定值完成的新的CompletableFuture。    return CompletableFuture.completedFuture(results);  }}

复制代码

测试编写的异步方法

@RestController@RequestMapping("/async")public class AsyncController {
  @Autowired   AsyncService asyncService;
  @GetMapping("/movies")  public String completableFutureTask() throws ExecutionException, InterruptedException {    //开始时间    long start = System.currentTimeMillis();    // 开始执行大量的异步任务    List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");    List<CompletableFuture<List<String>>> completableFutureList = words.stream()            .map(word -> asyncService.completableFutureTask(word))            .collect(Collectors.toList());    // CompletableFuture.join() 方法可以获取他们的结果并将结果连接起来    List<List<String>> results = completableFutureList.stream().map(CompletableFuture::join).collect(Collectors.toList());    // 打印结果以及运行程序运行花费时间    System.out.println("Elapsed time: " + (System.currentTimeMillis() - start));    return results.toString();  }}

复制代码

请求这个接口，控制台打印出下面的内容：

2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-1] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-1start this task!2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-6] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-6start this task!2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-5] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-5start this task!2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-4] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-4start this task!2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-3] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-3start this task!2019-10-01 13:50:17.007  WARN 18793 --- [lTaskExecutor-2] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-2start this task!Elapsed time: 1010

复制代码

首先我们可以看到处理所有任务花费的时间大概是 1 s。这与我们自定义的 ThreadPoolTaskExecutor 有关，我们配置的核心线程数是 6 ，然后通过通过下面的代码模拟分配了 6 个任务给系统执行。这样每个线程都会被分配到一个任务，每个任务执行花费时间是 1 s ，所以处理 6 个任务的总花费时间是 1 s。

List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");  List<CompletableFuture<List<String>>> completableFutureList =        words.stream()            .map(word -> asyncService.completableFutureTask(word))            .collect(Collectors.toList());

复制代码

试着去把核心线程数的数量改为 3 ，再次请求这个接口你会发现处理所有任务花费的时间大概是 2 s。

特殊情况无需返回值

另外，从上面的运行结果可以看出，当所有任务执行完成之后才返回结果。这种情况对应于我们需要返回结果给客户端请求的情况下，假如我们不需要返回任务执行结果给客户端的话呢？就比如我们上传一个大文件到系统，上传之后只要大文件格式符合要求我们就上传成功。普通情况下我们需要等待文件上传完毕再返回给用户消息，但是这样会很慢。采用异步的话，当用户上传之后就立马返回给用户消息，然后系统再默默去处理上传任务。这样也会增加一点麻烦，因为文件可能会上传失败，所以系统也需要一点机制来补偿这个问题，比如当上传遇到问题的时候，发消息通知用户。

下面会演示一下客户端不需要返回结果的情况：

将 completableFutureTask 方法变为 void 类型

@Asyncpublic void completableFutureTask(String start) {  ......  //这里可能是系统对任务执行结果的处理，比如存入到数据库等等......  //doSomeThingWithResults(results);}

复制代码

Controller 代码修改如下：

  @GetMapping("/movies")  public String completableFutureTask() throws ExecutionException, InterruptedException {    // Start the clock    long start = System.currentTimeMillis();    // Kick of multiple, asynchronous lookups    List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");        words.stream()            .forEach(word -> asyncService.completableFutureTask(word));    // Wait until they are all done    // Print results, including elapsed time    System.out.println("Elapsed time: " + (System.currentTimeMillis() - start));    return "Done";  }

复制代码

请求这个接口，控制台打印出下面的内容：

Elapsed time: 02019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-4] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-4start this task!2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-3] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-3start this task!2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-2] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-2start this task!2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-1] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-1start this task!2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-6] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-6start this task!2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-5] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-5start this task!

复制代码

可以看到系统会直接返回给用户结果，然后系统才真正开始执行任务。

下一章内容：

Future vs. CompletableFuture 的对比分析
CompetableFuture 和 Future 的源代码分析

参考引用

https://spring.io/guides/gs/async-method/
https://medium.com/trendyol-tech/spring-boot-async-executor-management-with-threadpooltaskexecutor-f493903617d

发布于: 5 小时前阅读数: 9

原文链接:【http://xie.infoq.cn/article/33f5542da83646fce2c9941e5】。文章转载请联系作者。

李浩宇/Alex

关注

🏆2021年InfoQ写作平台-签约作者 🏆 2020.03.25 加入

👑【酷爱计算机技术、醉心开发编程、喜爱健身运动、热衷悬疑推理的”极客狂人“】 🏅 【Java技术领域，MySQL技术领域，APM全链路追踪技术及微服务、分布式方向的技术体系等】我们始于迷惘，终于更高水平的迷惘

发布

暂无评论

创作场景

👊 【Spring 技术原理】异步编程机制以及功能分析讲解

本文内容

Future 模式

Future 的核心思想

SpringBoot 异步编程实战

TaskExecutor

自定义 AsyncConfigurer

特殊情况 无需返回值

下一章内容：

参考引用

李浩宇/Alex

评论

特殊情况无需返回值