【Spring Boot相关】探讨Spring mvc利用CompletableFuture实现的异步调用

【参考】

【本文内容】

  • Spring mvc利用CompletableFuture实现异步调用，探讨了以下问题：
    • 对比同步和异步单个请求的处理时间：这个异步是纯后端的异步，对前端是无感的，异步也并不会带来响应时间上的优化，原来该执行多久照样还是需要执行多久。
    • 在Controller中使用CompletableFuture，一定会提高效率吗？通过ab压力测试显示，不一定。
    • 同步请求 vs 异步请求所暂用的资源：异步处理暂用更多的资源。
  • CompletableFuture的默认线程池存在的问题：原因是ForkJoinPool默认的并发数太小。
  • 通过创建AOP，查看异步状态下的线程详细情况。

1. Spring mvc 异步请求 vs 同步请求

通常情况下Spring Controller返回请求数据的API长这样：

    @GetMapping("/sync_result")
    public String getResultSync() {
        sleep(500L);
        return "result is ready";
    }

但Spring也支持异步的调用，如返回Callable，或方法上加@Async，或返回CompletableFuture。这个功能实际上是Servlet 3.0之于异步请求的支持的实现。

例如返回CompletableFuture：

    @GetMapping("/async_result")
    public CompletableFuture<String> getResultAsyc() {
        return CompletableFuture.supplyAsync(() -> {
            sleep(500L);
            return "result is ready";
        });
    }
1.1 对比同步和异步单个请求的处理时间

这个异步是纯后端的异步，对前端是无感的，异步也并不会带来响应时间上的优化，原来该执行多久照样还是需要执行多久。

下图为F12的debug截图，可以看到对于单个请求，无论是同步的请求，还是异步的请求（使用CompletableFuture或Callable），都没有明显的差距：

image.png

1.2 在Controller中使用CompletableFuture，一定会提高效率吗？

默认情况下，Tomcat 的核心线程是 10，最大线程数是 200, 我们能及时回收线程，也就意味着我们能处理更多的请求，能够增加我们的吞吐量，这也是异步 Servlet 的主要作用。

那么使用CompletableFuture一定会提高吞吐量吗？

这里我们用apache bench做实验，关于ab，请参考：https://www.jianshu.com/p/557df0992795

测试两个case：

  • 同步请求数据，API为：/sync_request
  • 利用CompletableFuture异步请求数据，但使用自定义的线程池，线程数为500，API为：/async_request/custom_pool 
    @GetMapping("/async_result/custom_pool")
    public CompletableFuture<String> getResultAsycWithCustomPool() {
        return CompletableFuture.supplyAsync(() -> {
            sleep(500L);
            return "result is ready";
        }, executorService);
    }

我们做如下请求测试：
使用ab压力测试工具进行100/200/400/800的并发量测试，方法内延时时间分别为：10ms/100ms/500ms：

ab命令为：

ab -k -n 2000 -c 100 localhost:8080/sync_result
ab -k -n 4000 -c 200 localhost:8080/sync_result
ab -k -n 8000 -c 400 localhost:8080/sync_result
ab -k -n 16000 -c 800 localhost:8080/sync_result

ab -k -n 2000 -c 100 localhost:8080/async_result/custom_pool
ab -k -n 4000 -c 200 localhost:8080/async_result/custom_pool
ab -k -n 8000 -c 400 localhost:8080/async_result/custom_pool
ab -k -n 16000 -c 800 localhost:8080/async_result/custom_pool

ab结果，表中的时间为每个request所需的时间（毫秒）：
image.png

转为图表（省略小数点）：
image.png

可以看到：

  • 并发用户数不高的情况下（如100-200个用户）时，同步请求和异步请求所需的时间差不多。
  • 而在高并发的情况下（如用户>=400）时，异步请求的优势较为明显。
1.3 同步请求 vs 异步请求所暂用的资源

以下是请求处理时间为500ms时（sleep time=500ms），不同用户并发数（100/200/400/800）下，并且请求为【同步】时的Jconsole：
image.png

以下是相同条件下，请求为【异步】时的Jconsole：
image.png

可以看到异步时占用的内存更多，线程数也更多，占用的cpu更高。

2. CompletableFuture的默认线程池存在的问题

在上述#1.2做压力测试的时，使用的是500个核心线程的线程池，那么如果我们不自定义线程池，而是使用CompletableFuture默认的线程池，效率会如何呢？

尝试请求处理时间为10ms时，使用默认线程池的表现：

ab -k -n 2000 -c 100 localhost:8080/async_result

结果每个request平均需要162.894ms！！！

而同步的状态下，#1.2的测试结果为16.368ms，使用500个核心线程池时异步状态下的结果为18.103ms。

为什么CompletableFuture在默认线程池下这么慢？
默认情况下，CompletableFuture使用的是ForkJoinPool.commonPool()，而ForkJoinPool默认支持的并发数为cpu-1：

System.out.println("Parallelism:" + ForkJoinPool.commonPool().getParallelism());
System.out.println("CPU: " + Runtime.getRuntime().availableProcessors());

在我的pc下，分别为7和8。即8核的情况下ForkJoinPool默认的可并行级别为7。

而7个并行级别显然是太低了，还不如同步。这也是为什么在使用CompletableFuture作为Controller返回值时，推荐加上线程池的原因。

3. 异步状态下的线程详细情况

代码借鉴了开头参考列表中的第1个github repository。

首先我们自定义一个注解类：

@Target({ ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface AsyncTimed {
}

为了更好的观察，为注解类加一个aop，需要引入aop包以及google的guava包：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>31.1-jre</version>
        </dependency>

@Pointcut定义的包名为AsyncTimed所在的包名：

@Slf4j
@Aspect
@Component
public class AsyncTimedAspect {

    @Pointcut("@annotation(com.async.aspect.AsyncTimed)")
    public void asyncTimedAnnotationPointcut() {
    }

    @Around("asyncTimedAnnotationPointcut()")
    public Object methodsAnnotatedWithAsyncTime(final ProceedingJoinPoint joinPoint) throws Throwable {
        return proceed(joinPoint);
    }

    private Object proceed(final ProceedingJoinPoint joinPoint) throws Throwable {
        final Object result = joinPoint.proceed();
        if (!isCompletableFuture(result)) {
            return result;
        }

        final String description = joinPoint.toString();
        final Stopwatch watch = Stopwatch.createStarted();

        String initiatingThread = Thread.currentThread().getName();
        return ((CompletableFuture<?>) result).thenApply(__ -> {
            String executingThread = Thread.currentThread().getName();
            watch.stop();
            log.info("description: {}", description);
            log.info("watch: {}", watch.toString());
            log.info("initiating thread: {}", initiatingThread);
            log.info("executing thread: {}", executingThread);
            return __;
        });
    }

    private boolean isCompletableFuture(Object result) {
        return CompletableFuture.class.isAssignableFrom(result.getClass());
    }
}

定义Service类：

@Service
public class MessageService {
    ExecutorService executorService = Executors.newFixedThreadPool(500);

    @AsyncTimed
    public CompletableFuture<String> getMessageWithDefaultPool() {
        return CompletableFuture.supplyAsync(() -> worker());
    }

    @AsyncTimed
    public CompletableFuture<String> getMessageWithCustomPool() {
        return CompletableFuture.supplyAsync(() -> worker(), executorService);
    }

    public String worker() {
        try {
            Thread.sleep(500L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "result is ready";
    }
}

Controller类：

@RestController
@RequestMapping("message")
public class MessageController {
    @Autowired
    private MessageService messageService;

    @GetMapping("/custom-pool")
    public CompletableFuture<String> getResultAsyc() {
        log.info("Try to execute request with custom pool: {}", Thread.currentThread().getName());
        return messageService.getMessageWithCustomPool();
    }

    @GetMapping("/default-pool")
    public CompletableFuture<String> getResultAsycWithDefaultPool() {
        log.info("Try to execute request with default pool: {}", Thread.currentThread().getName());
        return messageService.getMessageWithDefaultPool();
    }
}

我们尝试调用API: http://localhost:8080/message/custom-pool：

image.png

可以看到在Controller层用的是http-nio线程池。而在返回结果的时候使用了我们自已定义的线程池。

而使用默认线程池时，返回结果用的是ForkJoinPool的commonPool：
image.png

如果我们并发40个请求数去访问，可以看到并发数最高还是7个：
image.png

而自定义核心线程数为500时，是可以充分利用线程池中的线程的：
image.png
尝试扩大ForkJoinPool的并发数
ForkJoinPool forkJoinPool = new ForkJoinPool(500);

这时候测试请求处理时间在500ms，并发数为800，结果为平均每个request花费时间为837.612ms。
而在#1.2中使用自定义核心线程数为500时，同样条件下的花费时间：836.737ms。

可以看到异步请求，瓶颈还是在线程池中的线程数。

4. 和WebFlux对比

参考：https://www.baeldung.com/spring-mvc-async-vs-webflux

加入依赖：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

写一个测试：

    @GetMapping("/flux_result")
    public Mono getResult() {
        Mono mono = Mono.defer(() -> Mono.just("Result is ready"))
                .delaySubscription(Duration.ofMillis(500));
        return mono;
    }

第一章#1.2的测试加上Webflux，可以看出在高延时、高并发的情况下，Webflux还是有优势的：
image.png

后续对Webflux，500ms延时，1000个用户和2000个用户下做了测试，平均每个request耗时分别是535.087ms和551.254ms，还是比较稳定的。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • 基于Spring框架实现异步请求与异步调用
    一、异步请求 1.1 同步请求与异步请求 首先看一下同步请求的线程执行模型： 接着看一下异步请求的线程执行模型： ...
    GeekerLou阅读 1,198评论 0赞 0
  • Spring Boot - 异步任务
    前言 有时候，前端可能提交了一个耗时任务，如果后端接收到请求后，直接执行该耗时任务，那么前端需要等待很久一段时间才...
    Whyn阅读 8,376评论 0赞 3
  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,958评论 19赞 139
  • Spring Framework 5 MVC 官方手册译文
    Spring Web MVC Spring Web MVC 是包含在 Spring 框架中的 Web 框架，建立于...
    Hsinwong阅读 22,574评论 1赞 92
  • SpringBoot线程池异步调用
    异步调用介绍 异步调用异步调用就是在不阻塞主线程的情况下执行高耗时方法 常规异步通过开启新线程实现 在Spring...
    weisen阅读 4,327评论 0赞 15