使用 Spring WebFlux 和 Reactor 的响应式 Java

Spring WebFlux 是 Java 中最流行的反应式编程框架之一。以下是结合使用 WebFlux 和 Reactor 的实践。

响应式编程是从函数世界演变而来的一种重要的编码风格。反应式代码利用流、生产者和订阅者的事件驱动原理来简化复杂的逻辑并支持应用程序中 IO 处理的异步、非阻塞处理。在 Java 中，这意味着我们可以使用java.nio具有表现力 API 的（非阻塞 IO）包来构建应用程序。许多框架和方法都支持 Java 中的反应性。最流行的之一是Spring WebFlux。本文是对使用 Spring WebFlux 进行响应式 Java 编程的实践介绍。

与 Spring 的反应性

反应性为我们提供了一个强大的单一惯用语来描述和组合 Web 请求和数据访问等功能。一般来说，我们使用事件生产者和订阅者来描述异步事件源以及事件被激活时应该发生的情况。

在典型的 Spring 框架风格中，WebFlux 提供了一个用于构建反应式 Web 组件的抽象层。这意味着您可以使用几种不同的底层响应式实现。默认为 Reactor，我们将使用它进行演示。

首先，我们将使用 Spring 命令行工具初始化一个新应用程序。有几种方法可以安装这个工具，但我喜欢使用 SDKMan。您需要安装 Java 17+。您可以在此处找到针对您的操作系统安装 SDKMan 的说明。安装后，您可以使用以下命令添加 Spring CLI $ sdk i springboot：现在该命令$ spring --version应该可以工作了。

要启动新应用程序，请键入：

$ spring init --dependencies=webflux --build=maven --language=java spring-reactive

接下来，cd进入spring-reactive目录。Spring 为我们创建了一个简单的布局，包括src/main/java/com/example/springreactive2/DemoApplication.java. 

让我们修改此类以添加反应式端点处理程序，如清单 1 所示。

清单 1. 添加 RESTful 端点

package com.example.springreactive;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestParam;

@SpringBootApplication
public class DemoApplication {
  public static void main(String[] args) {
    SpringApplication.run(DemoApplication.class, args);
  }
  @RestController
  public class EchoController {
    @GetMapping("/hello")
    public Mono<String> hello() {
      return Mono.just("Hello, InfoWorld!");
    }
    @GetMapping("echo/{str}")
    public Mono<String> echo(@PathVariable String str) {
      return Mono.just("Echo: " + str);
    }
    @GetMapping("echoquery")
    public Mono<String> echoQuery(@RequestParam("name") String name) {
      return Mono.just("Hello, " + name);
    }
  }
}

注意：在 Reactor 库中，Mono是指“一元值”的类型。

他们@SpringBootApplication为我们处理了大部分配置。我们使用一个EchoController通过@RestController注释调用的内部类，让 Spring 知道我们正在使用哪些端点。 

清单 1 中有三个示例，每个示例都映射到一个带有@GetMapping. 第一个，/hello简单地在响应中写入问候语。第二个/echo/{str}演示如何获取 URL 参数（路径变量）并在响应中使用它。第三个，/echoquery展示了如何获取请求参数（URL 中问号后面的值）并使用它。

在每种情况下，我们都依靠Mono.just()方法来描述响应。这是在 Reactor 框架中创建事件生成器的简单方法。它说：使用参数中找到的单个事件创建一个事件生成器，并将其交给所有订阅者。在这种情况下，订阅者由 Spring WebFlux 框架和托管它的非阻塞服务器处理。简而言之，我们可以访问完全非阻塞的响应管道。

入站请求处理也完全基于非阻塞 IO。这使得扩展服务器可能非常高效，因为没有阻塞线程来限制并发性。特别是在实时系统中，非阻塞 IO 对于整体吞吐量非常重要。

Spring WebFlux默认使用Netty 服务器。如果您愿意，可以使用其他服务器（例如 Undertow）或 Servlet 3.1 容器（例如 Tomcat）。有关服务器选项的更多信息，请参阅 WebFlux 文档。

反应式编程示例

响应式编程需要一整套思维方式和一组概念，我们在这里不会对其进行探讨。相反，我们将通过一些示例来揭示这种编程风格的关键方面。

首先，让我们创建一个接受文件上传帖子并将内容写入磁盘的端点。您可以在清单 2 中看到该方法及其导入。代码的其余部分保持不变。

清单 2. 接受并写入文件

import org.springframework.http.MediaType; 
import org.springframework.http.codec.multipart.FilePart; 
import org.springframework.web.bind.annotation.PostMapping; 
import org.springframework.web.bind.annotation.RequestPart; 
import org.springframework.web.bind.annotation.RestController; 
import org.springframework.util.FileSystemUtils; 
import reactor.core.publisher.Flux; 
import reactor.core.publisher.Mono;

@PostMapping(value = "/writefile", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<String> writeFile(@RequestPart("file") Flux<FilePart> filePartFlux) {
        Path path = Path.of("/tmp/file.txt");

        // Delete the existing file if it already exists
        FileSystemUtils.deleteRecursively(path.toFile());

        // Save the file parts to the specified path
        return filePartFlux
                .flatMap(filePart -> filePart.transferTo(path))
                .then(Mono.just("File saved: " + path.toString()));
    }

该writeFile()方法用 注释@PostMapping，并且配置为接受分段表单上传。到目前为止，这是一个正常的 Spring Web 配置。WebFlux 允许我们在方法参数中使用@RequestPart类型为 的注释Flux<FilePart>。这让我们可以使用 Flux 以反应式、非阻塞的方式接受多部分块。

有了这个filePartFlux，我们就可以使用反应式flatMap方法将其写入磁盘：filePartFlux.flatMap(filePart -> filePart.transferTo(path))。多部分文件的每个“事件”都会传递给transferTo要添加到文件中的函数。这是一个非常惯用的反应操作。

使用高阶函数（例如flatMap转换和处理事件流）是反应式编程的典型。它可以被视为具有事件生产者、订阅者和转换器，例如flatMap. 通过获取一个或多个流并使用元函数链对其进行操作，您可以使用相对简单的语法实现强大的效果。

要测试新端点，您可以使用 CURL 命令，如清单 3 所示。

清单 3. 使用 CURL 测试 writeFile 端点

$ echo "Facing it, always facing it, that’s the way to get through." >> testfile.txt
$ curl -X POST -F "file=@./testfile.txt" http://localhost:8080/writefile
File saved: /tmp/file.txtmatthewcarltyson@dev3:~/spring-reactive2

$ cat /tmp/file.txt

在清单 3 中，我们创建一个testfile.txt包含一些内容的文件（“面对它，始终面对它，这就是通过的方法”），然后将其发送到端点，接收响应，并验证新文件内容。

使用反应式 HTTP 客户端

现在，让我们创建一个接受 ID 参数的端点。我们将使用 Spring 反应式 HTTP 客户端在SWAPI （星球大战 API）上向该 ID 处的角色发出请求，然后将角色数据发送回用户。您可以在清单 4 中看到这个新apiChain()方法及其导入。

清单 4. 使用响应式 Web 客户端

import org.springframework.web.reactive.function.client.WebClient;

@GetMapping("character/{id}")
  public Mono<String> getCharacterData(@PathVariable String id) {
    WebClient client = WebClient.create("https://swapi.dev/api/people/");
    return client.get()
      .uri("/{id}", id)
      .retrieve()
      .bodyToMono(String.class)
      .map(response -> "Character data: " + response);
  }

现在，如果您导航到localhost:8080/character/10，您将获得欧比旺·克诺比的传记信息。

清单 4 的工作方式是接受 ID 路径参数并使用它向类发出请求WebClient。在本例中，我们正在为请求创建一个实例，但您可以WebClient使用基本 URL 创建一个实例，然后在许多路径中重复使用它。我们将 ID 放入路径中，然后调用retrieve，bodyToMono()将响应转换为 Mono。请记住，这一切仍然是非阻塞和异步的，因此等待 SWAPI 响应的代码不会阻塞线程。最后，我们用来map()制定返回给用户的响应。

所有这些示例的总体效果是启用从服务器一直到代码的高性能堆栈，而无需大惊小怪。

结论

响应式编程是一种与更熟悉的声明式风格不同的思维方式，这可能会使推理简单场景变得更加困难。找到理解反应式编程的程序员也更难。一般来说，良好的技术或业务需求应该决定您是使用反应式框架还是更标准的框架。