Spring Boot 3.4 RestClient同步调用在虚拟线程中阻塞Carrier线程调度？RestClientBuilder.requestFactory与VirtualThreadPerTaskExecutor

Spring Boot 3.4 RestClient同步调用与虚拟线程：阻塞Carrier线程的调度？

大家好！今天我们来探讨一个在Spring Boot 3.4中使用RestClient进行同步调用时，结合虚拟线程可能遇到的一个比较棘手的问题：同步调用阻塞Carrier线程，进而影响虚拟线程的调度。这个问题涉及到Java虚拟线程、Spring RestClient、阻塞IO以及线程调度等多个方面，理解其原理对于编写高性能、可扩展的应用程序至关重要。

1. 虚拟线程（Virtual Threads）简介

首先，我们需要简单了解一下Java的虚拟线程。虚拟线程是Java 21引入的一项重要特性，旨在大幅度降低线程的创建和切换成本，从而提高并发程序的性能。与传统的平台线程（Platform Threads，也称操作系统线程）相比，虚拟线程具有以下显著特点：

轻量级： 虚拟线程的创建和销毁成本极低，可以创建数百万个虚拟线程而不会耗尽系统资源。
用户态线程： 虚拟线程完全由JVM管理，其调度和切换发生在用户态，避免了频繁的内核态切换开销。
多路复用： 多个虚拟线程可以多路复用到少量的平台线程（称为Carrier线程）上执行。当虚拟线程执行阻塞IO操作时，它会“卸载”到Carrier线程，允许Carrier线程执行其他虚拟线程。

这种多路复用机制是虚拟线程实现高并发的关键。当一个虚拟线程阻塞时，Carrier线程能够迅速切换到另一个就绪的虚拟线程，避免了整个线程的阻塞，从而提高了系统的吞吐量。

2. Spring RestClient简介

Spring RestClient是Spring Framework 5.1引入的用于进行RESTful API调用的客户端工具。它提供了一种简洁、灵活的方式来发送HTTP请求并处理响应。相比于RestTemplate，RestClient提供了更现代化的API和更强的可扩展性。

在Spring Boot 3.4中，我们可以通过RestClientBuilder来构建RestClient实例。RestClientBuilder允许我们配置各种选项，例如请求拦截器、响应处理器、错误处理器等。

3. 问题描述：同步调用阻塞Carrier线程

现在，我们来描述一下今天讨论的核心问题。当我们在虚拟线程中使用RestClient进行同步调用时，如果RESTful API调用发生阻塞（例如，网络延迟、服务器响应缓慢），那么Carrier线程可能会被阻塞，从而影响其他虚拟线程的调度。

为什么会出现这个问题？

这是因为RestClient默认情况下使用阻塞IO。当RestClient发起一个同步HTTP请求时，它会阻塞当前线程，直到收到响应或发生超时。如果在虚拟线程中执行此操作，那么Carrier线程就会被阻塞，导致其他等待执行的虚拟线程无法被调度。

4. 代码示例

为了更清楚地说明这个问题，我们来看一个简单的代码示例。

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.client.RestClientBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestClient;

import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.stream.IntStream;

@SpringBootApplication
public class VirtualThreadRestClientApplication {

    public static void main(String[] args) {
        SpringApplication.run(VirtualThreadRestClientApplication.class, args);
    }

    @RestController
    static class MyController {

        private final RestClient restClient;

        public MyController(RestClientBuilder restClientBuilder) {
            this.restClient = restClientBuilder.baseUrl("https://httpstat.us").build(); // Use a service that can simulate delays
        }

        @GetMapping("/test")
        public String testVirtualThreads() throws InterruptedException {
            ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

            IntStream.range(0, 10).forEach(i -> {
                executor.submit(() -> {
                    System.out.println("Thread " + i + " started");
                    String result = restClient.get()
                            .uri("/200?sleep=500") // Simulate 500ms delay
                            .retrieve()
                            .body(String.class);
                    System.out.println("Thread " + i + " finished: " + result);
                    return result;
                });
            });

            executor.shutdown();
            executor.awaitTermination(10, Duration.SECONDS);
            return "Test started";
        }
    }
}

在这个示例中，我们创建了一个RestController，它使用RestClient调用https://httpstat.us/200?sleep=500这个URL。这个URL会模拟500毫秒的延迟。我们使用Executors.newVirtualThreadPerTaskExecutor()创建了一个虚拟线程池，并提交了10个任务到线程池中。每个任务都会调用RestClient进行同步HTTP请求。

如果你运行这段代码，你会发现，尽管我们使用了虚拟线程，但程序的执行速度并没有显著提升。这是因为每个虚拟线程的同步调用都会阻塞Carrier线程，导致其他虚拟线程无法及时被调度。

5. RestClientBuilder.requestFactory与VirtualThreadPerTaskExecutor

那么，如何解决这个问题呢？关键在于使用非阻塞IO。我们可以通过配置RestClientBuilder的requestFactory属性来实现。

requestFactory属性允许我们指定RestClient使用的ClientHttpRequestFactory。ClientHttpRequestFactory负责创建ClientHttpRequest对象，而ClientHttpRequest对象负责实际发送HTTP请求。

Spring提供了多种ClientHttpRequestFactory的实现，其中一些支持非阻塞IO，例如：

ReactorNettyClientRequestFactory： 使用Reactor Netty作为HTTP客户端，提供非阻塞IO支持。
JettyClientRequestFactory: 使用Jetty HttpClient作为HTTP客户端，提供非阻塞IO支持.

我们可以使用ReactorNettyClientRequestFactory来配置RestClientBuilder，从而实现非阻塞IO。

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.client.RestClientBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.http.client.ReactorNettyClientRequestFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestClient;

import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.stream.IntStream;

@SpringBootApplication
public class VirtualThreadRestClientApplication {

    public static void main(String[] args) {
        SpringApplication.run(VirtualThreadRestClientApplication.class, args);
    }

    @Bean
    public RestClientBuilder restClientBuilder() {
        return RestClient.builder()
                .requestFactory(new ReactorNettyClientRequestFactory());
    }

    @RestController
    static class MyController {

        private final RestClient restClient;

        public MyController(RestClientBuilder restClientBuilder) {
            this.restClient = restClientBuilder.baseUrl("https://httpstat.us").build(); // Use a service that can simulate delays
        }

        @GetMapping("/test")
        public String testVirtualThreads() throws InterruptedException {
            ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

            IntStream.range(0, 10).forEach(i -> {
                executor.submit(() -> {
                    System.out.println("Thread " + i + " started");
                    String result = restClient.get()
                            .uri("/200?sleep=500") // Simulate 500ms delay
                            .retrieve()
                            .body(String.class);
                    System.out.println("Thread " + i + " finished: " + result);
                    return result;
                });
            });

            executor.shutdown();
            executor.awaitTermination(10, Duration.SECONDS);
            return "Test started";
        }
    }
}

在这个修改后的示例中，我们添加了一个@Bean来配置RestClientBuilder，并使用ReactorNettyClientRequestFactory作为requestFactory。这样，RestClient就会使用Reactor Netty进行非阻塞IO。

现在，如果你再次运行这段代码，你会发现程序的执行速度显著提升。这是因为Reactor Netty使用非阻塞IO，当一个虚拟线程的HTTP请求发生阻塞时，Carrier线程不会被阻塞，而是可以继续执行其他虚拟线程。

6. 深入理解：阻塞IO与非阻塞IO

为了更好地理解这个问题，我们需要深入了解一下阻塞IO与非阻塞IO的区别。

阻塞IO：

当一个线程发起IO操作时，如果数据尚未准备好，线程会被阻塞，直到数据准备好或发生超时。
在阻塞期间，线程无法执行其他任务。
适用于IO操作不频繁的场景。

非阻塞IO：

当一个线程发起IO操作时，如果数据尚未准备好，线程不会被阻塞，而是立即返回一个状态码，表示IO操作尚未完成。
线程可以继续执行其他任务，并在稍后再次检查IO操作是否完成。
适用于IO操作频繁的场景。

在使用虚拟线程时，我们应该尽可能使用非阻塞IO，以避免阻塞Carrier线程，从而充分发挥虚拟线程的优势。

7. 其他考虑因素

除了使用非阻塞IO之外，还有一些其他的因素需要考虑：

连接池： 使用连接池可以减少HTTP连接的创建和销毁开销，提高性能。Reactor Netty和Jetty HttpClient都提供了连接池功能。
超时设置： 设置合理的超时时间可以防止HTTP请求无限期地阻塞，从而避免阻塞Carrier线程。
异步编程模型： 考虑使用异步编程模型（例如，使用WebClient或CompletableFuture）来进一步提高并发性能。

8. 表格总结：解决方案与适用场景

解决方案	描述	适用场景	优点	缺点
使用非阻塞IO (Reactor Netty)	配置RestClientBuilder使用ReactorNettyClientRequestFactory。	高并发，频繁IO操作，需要充分利用虚拟线程的场景。	避免阻塞Carrier线程，提高并发性能，充分发挥虚拟线程的优势。	需要引入Reactor Netty依赖，增加项目复杂度。 Reactor Netty的学习曲线相对较陡峭。需要熟悉响应式编程模型。
使用连接池	配置Reactor Netty或Jetty HttpClient的连接池参数。	高并发，需要频繁建立和关闭HTTP连接的场景。	减少HTTP连接的创建和销毁开销，提高性能。	需要合理配置连接池参数，避免连接泄漏或连接耗尽。
设置合理的超时时间	配置RestClient的超时时间参数。	所有场景，特别是网络不稳定或服务器响应缓慢的场景。	防止HTTP请求无限期地阻塞，避免阻塞Carrier线程。	需要根据实际情况设置合理的超时时间，过短的超时时间可能导致请求失败。
使用异步编程模型 (WebClient)	使用WebClient进行非阻塞的异步HTTP请求。	高并发，需要最大化利用系统资源的场景。	进一步提高并发性能，充分利用系统资源。	需要熟悉异步编程模型，代码复杂度较高。

9. 虚拟线程优势的发挥需要非阻塞的配合

总结一下，在Spring Boot 3.4中使用RestClient进行同步调用时，如果结合虚拟线程，需要特别注意阻塞IO可能导致的问题。为了充分发挥虚拟线程的优势，我们应该尽可能使用非阻塞IO，例如通过配置RestClientBuilder的requestFactory属性，使用ReactorNettyClientRequestFactory或JettyClientRequestFactory。此外，还需要考虑连接池、超时设置和异步编程模型等因素，以进一步提高并发程序的性能和可扩展性。

希望今天的分享对大家有所帮助！谢谢大家！