转载

Spring 5 响应式编程

近年来，响应式编程在开发者社区和客户中很受欢迎，由于其以声明的方式构建应用程序的能力，而不是强制，形成更加敏感和有弹性的应用。Spring 5 将反应系统纳入其核心框架的事实已经显示出向声明式编程的范式转变。

响应式编程管理数据生产者和消费者之间的异步数据流，它们需要以流畅的方式对数据进行响应。 所以，响应式编程 都是异步和事件驱动的流畅应用程序，需要少量的线程进行缩放。

响应式编程很难构建基于线程的架构，由于在基于共享可变状态、线程和锁的应用程序扩展过程中涉及到高度复杂性。

在响应式编程的上下文中，“在流中有数据时，所有事物都是流并以流畅方式行为。”

为什么是响应式编程

高层次的抽象与响应式编程导致了代码可读性的提高，因此开发人员可以主要关注定义业务逻辑的事件的相互依存性。

在高度并发的环境中，响应模式自然地适合于消息处理，这是一个常见的企业用例。

由于执行反压力的特性，响应式方法最适合控制生产者和消费者之间的流量，这将有助于避免内存不足。

对于一个或几个线程，IO绑定任务可以通过异步和非阻塞方式执行，而且不阻塞当前线程。

在高交互和实时应用程序或任何操作/事件时，都可能触发多个连接子系统的通知，在这种情况下响应式编程可以更有效的进行管理。

用于响应式编程实现的理想案例

大量的交易处理服务，如银行部门。
大型在线购物应用程序的通知服务，如亚马逊。
股票价格同时变动的股票交易业务。

响应流 (Reactive Streams)

“响应流”定义一个 API 规范，其中包含一组最小的接口，这些接口公开了定义非阻塞反压力的数据流的操作和实体的方法。

随着反压力的引入，响应流允许用户控制来自出版商的数据交换速率。

响应流 API java.util.concurrent.flow 已正式成为 java 9 的一部分。

响应流主要用作互操作层。

Spring 5 提供的响应式编程

Spring-Web-Reactive 模块和 Spring MVC 都支持相同的 @Controller 编程，但 Spring-Web-Reactive 是在 Reactive 和非阻塞引擎上执行的。

Spring-Web-Reactive 模块和 Spring MVC 共享许多通用的算法，但 Spring-Web-Reactive 模块已经重新定义了许多 Spring MVC 契约，例如 HandlerMapping和HandlerAdapter，以使它们变得异步和非阻塞，以及启用响应式 HTTP 请求和响应（以 RouterFunction 和 HandlerFunction 的形式）。

除了现有的 RestTemplate 外，新的响应式 WebClient 也在 Spring 5 中引入。

支持响应式编程的 HTTP 客户端（例如 Reactor、Netty、Undertow）已经采用了一系列响应式的 ClientHttpRequest 和 ClientHttpResponse 抽象，它们将请求和响应体作为 Flux <DataBuffer> 公开，并且在读写端提供全面的背压支持。

Spring 5 Framework 引入了 Reactor， 作为 Reactive Streams 规范的实现。

Reactor 是用于在 JVM 上构建非阻塞应用程序的下一代 Reactive 库。

Reactor 扩展了基本的 Reactive Streams Publisher 契约，并定义了 Flux 和 Mono API 类型，分别为 0..N 和 0..1 的数据序列提供声明性操作。

Spring Web Reactive 使用 Servlet 3.1 提供了非阻塞 I/O，并运行在 Servlet 3.1 容器上。

Spring WebFlux 提供了两种编程模型可供选择。

注解式控制器：这些与 Spring MVC 相同，在使用一些 Spring-Web 模块提供的附加注解的情况下。Spring MVC 和 WebFlux 控制器都支持 Reactive 返回类型。另外，WebFlux 还支持 Reactive @RequestBody 参数。
函数式编程模型：基于 lambda 的轻量级小型库，用于给出路由和处理请求的实用程序。

Spring Web Reactive vs. Spring Web MVC

Spring 5 包含了 Spring Web Reactive（在 spring-web-reactive 模块下）和 Spring Web MVC（在 spring-webmvc 模块下）。

虽然 Spring Web Reactive 和 Spring Web MVC 模块共享了很多算法，但是由于 Spring Web Reactive 能够在响应式的非阻塞 Reactive Streams HTTP 适配器层上运行，因此它们之间不共享代码。

Spring MVC 执行需要 Servlet 容器，而 Spring Web Reactive 也可以在非 Servlet 运行时上运行，比如在 Netty 和 Undertow 上运行。

如果对具有轻量级、函数式的 Web 框架的非阻塞 Web 栈的必要要求，并与 Java 8 lambda 或 Kotlin一起使用，则应该考虑从 Spring MVC 应用程序切换到 Spring Web Reactive。

响应式编程的基本结构

这是一个 2.0.0 M5 版本的 pom.xml ，依赖于 WebFiux。

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.0.M5</version>
</parent>
<dependencies>         
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>                               </dependency>
</dependencies>

传统方法 VS. 响应方法

在传统方法中，执行将被阻塞，并将一直等到您的服务执行完成为止。在下面的代码中，在第一个 print 语句之后，程序执行将 被阻塞并等待 服务执行完成。在服务执行完成后，将恢复程序执行，并执行第二个 print 语句。

@GetMapping("/traditional")
public List < Product > getAllProducts() {
    System.out.println("Traditional way started");
    List < Product > products = prodService.getProducts("traditional");
    System.out.println("Traditional way completed");
    return products;
}

在响应方法中，程序执行将继续，而不用等待服务执行的完成。在下面的代码中，在第一个 print 语句之后，第二个 print 语句将以非阻塞的方式执行，而无需等待服务执行的完成。Flux stream 将随着产品数据的可用性而被填充。

@GetMapping(value = "/reactive", .TEXT_EVENT_STREAM_VALUE)
public Flux < Product > getAll() {
    System.out.println("Reactive way using Flux started");
    Flux < Product > fluxProducts = prodService.getProductsStream("Flux");
    System.out.println("Reactive way using Flux completed");
    return fluxProducts;
}

Reactive Web Client

除了现有的 RestTemplate 之外，Spring 5 还引入了 Reactive WebClient。

ClientHttpRequest 和 ClientHttpResponse 抽象将请求和响应主体使用 Flux <DataBuffer> 公开，并在读和写端提供完全背压支持。

来自 Spring Core 的编码器和解码器抽象也用在客户端，用于将字节序的 Flus 序列化或发序列化成类型对象。

下面是一个 Reactive WebClient 的示例，它调用终端并接收及处理 Reactive Stream Flux 对象。

@GetMapping("/accounts/{id}/alerts")
public Flux < Alert > getAccountAlerts(@PathVariable Long id) {
    WebClient webClient = new WebClient(new ReactorClientHttpConnector());
    return this.repository.getAccount(id).flatMap(account -> webClient.perform(get("/alerts/{key}", account.getKey())).extract(bodyStream(Alert.class)));
}