Spring Reactor教程

在RESTful服务的世界中，实际上实际上是在幕后进行许多工作，我们通常必须在应用程序中进行很多处理，而实际上并不会影响需要发送给真实用户的响应。可以被动地做出这些业务决策，以便它们对与应用程序交互的用户没有任何影响。 Spring Framework为我们提供了一个出色的项目，称为Spring Reactor项目，它使我们能够在后台很好地管理此后台处理。在学习本课程之前，我们必须注意一件事，即反应式编程与并发编程并不相同 。

RESTful编程中用于响应行为的用例之一是，在大多数情况下，服务从根本上是阻塞和同步的。响应式编程使我们可以扩展到同步线程的范围之外，并且可以在不表现阻塞行为的情况下完成复杂的编排。让我们深入学习本课程，以了解如何将这种反应性行为集成到基于Spring Boot的应用程序中。

1.简介

在本Spring Reactor课程中，我们将学习如何在Spring Boot项目中开始反应性行为，以及如何在同一应用程序本身中开始产生和使用消息。除了一个简单的项目外，当有多个不同类型的请求处理程序时，我们还将看到Spring Reactive流如何工作以及如何管理请求。

随着的起义微服务，涉及的服务之间的异步通信的必要性成为主流需求。为了在涉及的各种服务之间进行通信，我们可以使用Apache Kafka之类的项目。现在，异步通信对于同一应用程序中的耗时请求也很理想。这是Spring Reactor的实际用例发挥作用的地方。

请注意，仅当用户不希望直接从应用程序获得响应时，才使用此应用程序中演示的Reactor模式，因为我们仅使用此Reactor演示执行后台作业。当开发人员可以为应用程序分配更多的堆内存（取决于该应用程序将使用的线程数）并且他们想并行执行任务并且任务的执行顺序不合理时，使用Reactor是一个很好的选择。没关系。这一点实际上很重要，因此我们将通过重新措辞再说一遍，当并行执行作业时，无法确认作业的执行顺序 。

2. JVM中的Reactor

正如Spring本身所言，Reactor是JVM上异步应用程序的基础框架，它在适度的硬件上，可以用最快的非阻塞Dispatcher 每秒处理超过15,000,000个事件 。听起来，Reactor框架基于Reactor设计模式。

关于Spring Reactor，最重要的是该框架为使用Spring开发应用程序的Java开发人员提供的抽象级别。这种抽象使得在我们自己的应用程序中实现功能非常容易。让我们从一个示例项目开始，看看如何在接近现实的应用程序中使用该框架。该反应堆项目还支持与反应堆IPC组件进行无阻塞的进程间通信（IPC），但其讨论不在本课程的讨论范围之内。

3.使用Maven制作Spring Boot项目

我们将使用许多Maven原型之一为我们的示例创建一个示例项目。要创建项目，请在将用作工作空间的目录中执行以下命令：

创建一个项目

mvn archetype:generate -DgroupId=com.javacodegeeks.example -DartifactId=JCG-BootReactor-Example -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

如果您是第一次运行maven，则完成生成命令将花费几秒钟，因为maven必须下载所有必需的插件和工件才能完成生成任务。运行该项目后，我们将看到以下输出并创建该项目：

Spring Reactor项目设置

4.添加Maven依赖项

创建项目后，请随时在您喜欢的IDE中打开它。下一步是向项目添加适当的Maven依赖关系。我们将在项目中使用以下依赖项：

spring-boot-starter-web ：此依赖关系将该项目标记为Web项目，并添加了依赖关系以创建控制器并创建与Web相关的类
reactor-bus ：这是将所有与Reactor相关的依赖项引入项目类路径的依赖项
spring-boot-starter-test ：此依赖项将所有与测试相关的JAR收集到项目中，例如JUnit和Mockito

这是pom.xml文件，其中添加了适当的依赖项：

pom.xml

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.10.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-bus</artifactId><version>2.0.8.RELEASE</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>

在Maven Central上找到最新的Maven依赖项。我们还为Spring引导项目添加了一个Maven插件，该插件可以帮助我们将该项目变成可运行的JAR，以便无需任何现代工具和依赖项即可轻松部署该项目。我们从此插件获得的JAR已完全准备好作为可执行文件进行部署。

最后，要了解添加此依赖项时添加到项目中的所有JAR，我们可以运行一个简单的Maven命令，当我们向其添加一些依赖项时，该命令使我们能够查看项目的完整依赖关系树。当我们以适当的层次结构方式添加一些自己的依赖项时，此依赖关系树还将显示添加了多少个依赖项。这是我们可以使用的相同命令：

检查依赖树

mvn dependency:tree

当我们运行此命令时，它将向我们显示以下依赖关系树：

Maven依赖树

注意到了什么？只需在项目中添加三个依赖项，即可添加如此多的依赖项。 Spring Boot本身会收集所有相关的依赖项，因此在此方面不做任何事情。最大的优点是，由于Spring Boot项目的pom文件本身可以管理和提供这些依赖关系，因此可以确保所有这些依赖关系相互兼容。

5.项目结构

在继续进行并开始为该项目编写代码之前，让我们介绍一下一旦完成将所有代码添加到项目中之后将拥有的项目结构，以便我们知道将在该项目中放置类的位置：

Spring Reactor项目结构

我们将项目分为多个包，以便遵循关注点分离的原则，并且代码保持模块化，这使得项目的扩展相当容易。

6.了解示例应用程序

为了使应用程序易于理解并且接近实际情况，我们将考虑一种物流应用程序的场景，该应用程序管理放置在系统中的各种货物的交付。

该应用程序从外部供应商处接收有关在给定地址处交付给客户的货件位置的更新。我们的应用程序收到此更新后，便会执行各种操作，例如：

在数据库中更新装运位置
向用户的移动设备发送通知
发送电子邮件通知
发送短信给用户

我们选择对这些操作表现出反应性行为，因为用户不依赖于这些操作是否实时准确地进行，因为它们主要是后台任务，这也可能会花费一些时间，并且如果装运状态更新晚了几分钟。让我们首先开始创建模型。

7.定义POJO模型

我们将从定义我们的POJO开始，该POJO表示要发送给客户的shipmentId ，该shipmentId具有currentLocation ， currentLocation等字段。让我们在这里定义此POJO：

Shipment.java

package com.javacodegeeks.example.model;public class Shipment {private String shipmentId;private String name;private String currentLocation;private String deliveryAddress;private String status;//standard setters and getters
}

我们在这里定义了一些基本字段。为了简洁起见，我们省略了标准的getter和setter方法，但是由于Jackson在对象的序列化和反序列化过程中使用它们，因此必须将它们制成。

8.定义服务

我们将定义一个基本接口，该接口定义我们接下来将要使用的功能的合同，该接口将定义一旦应用程序消耗了偶数就需要执行的业务逻辑。

这是我们将使用的合同定义：

ShipmentService.java

package com.javacodegeeks.example.service;import com.javacodegeeks.example.model.Shipment;public interface ShipmentService {void shipmentLocationUpdate(Shipment shipment);
}

在此接口中，我们只有一个方法定义，因为这是我们现在所需要的。现在让我们继续实现此服务，在这里我们将实际演示一个sleep方法，该方法只是模拟此类的操作行为：

ShipmentServiceImpl.java

package com.javacodegeeks.example.service;import com.javacodegeeks.example.model.Shipment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;@Service
public class ShipmentServiceImpl implements ShipmentService {private final Logger LOG = LoggerFactory.getLogger("ShipmentService");@Overridepublic void shipmentLocationUpdate(Shipment shipment) throws InterruptedException {LOG.info("Shipment data: {}", shipment.getShipmentId());Thread.sleep(3000);LOG.info("Shipment with ID: {} reached at javacodegeeks!!!", shipment.getShipmentId());}
}

出于说明目的，在调用此服务并附带装运详细信息时，它仅提供一些打印语句，使用3000毫秒的延迟来实现我们在上一节中定义的操作可能要花费的时间。请注意，这些操作中的每一个可能花费的时间远远超过3秒，但是应用程序没有时间（直到线程开始堆积在需要管理的应用程序的堆内存上）。

9.定义事件使用者

在本节中，我们最终将看到如何定义一个侦听事件发运位置更新的使用者。可以通过将事件进行装运更新来调用此使用者，该事件将在我们即将定义和使用的SPring的EventBus上放置。

EventHandler.java

package com.javacodegeeks.example.handler;import com.javacodegeeks.example.model.Shipment;
import com.javacodegeeks.example.service.ShipmentService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.fn.Consumer;@Service
public class EventHandler implements Consumer<Event<Shipment>> {private final ShipmentService shipmentService;@Autowiredpublic EventHandler(ShipmentService shipmentService) {this.shipmentService = shipmentService;}@Overridepublic void accept(Event<Shipment> shipmentEvent) {Shipment shipment = shipmentEvent.getData();try {shipmentService.shipmentLocationUpdate(shipment);} catch (InterruptedException e) {//do something as bad things have happened}}
}

此使用者服务在事件总线中接受该对象，并通知我们的服务类，以便它可以异步执行必要的操作。请注意，我们还将定义一个线程池，该线程池将用于运行此使用者，以便可以使用不同的线程来运行服务方法调用。即使我们自己没有定义线程池，Spring Boot也会使用固定数量的最大线程池为我们完成此任务。

此消费者类的好处是，它从事件总线接收到了Shipment对象本身，并且无需在类本身中进行转换或强制转换，这是常见的错误区域，并且还增加了业务逻辑所需的时间执行。

10.定义Java配置

我们可以在应用程序中使用Java定义配置。让我们在这里做这些定义：

ReactorConfig.java

package com.javacodegeeks.example.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.Environment;
import reactor.bus.EventBus;@Configuration
public class ReactorConfig {@BeanEnvironment env() {return Environment.initializeIfEmpty().assignErrorJournal();}@BeanEventBus createEventBus(Environment env) {return EventBus.create(env, Environment.THREAD_POOL);}
}

显然，这里没有什么特别的。我们只是用一些数字（这里是默认值）初始化了线程池。我们只是想演示如何根据您的应用程序用例来更改线程数。

11.定义Spring Boot类

在最后阶段，我们将创建Spring Boot类，通过该类我们可以发布一条消息，该消息可以由我们先前定义的事件处理程序使用。这是主类的类定义：

应用程序

package com.javacodegeeks.example;import com.javacodegeeks.example.handler.EventHandler;
import com.javacodegeeks.example.model.Shipment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import reactor.bus.Event;
import reactor.bus.EventBus;import static reactor.bus.selector.Selectors.$;@SpringBootApplication
public class Application implements CommandLineRunner {private final Logger LOG = LoggerFactory.getLogger("Application");private final EventBus eventBus;private final EventHandler eventHandler;@Autowiredpublic Application(EventBus eventBus, EventHandler eventHandler) {this.eventBus = eventBus;this.eventHandler = eventHandler;}public static void main(String[] args) {SpringApplication.run(Application.class, args);}@Overridepublic void run(String... strings) throws Exception {eventBus.on($("eventHandler"), eventHandler);//Publish messages herefor (int i = 0; i < 10; i++) {Shipment shipment = new Shipment();shipment.setShipmentId(String.valueOf(i));eventBus.notify("eventHandler", Event.wrap(shipment));LOG.info("Published shipment number {}.", i);}}
}

我们使用了CommandLineRunner接口来使此类运行代码，从而可以测试所编写的生产者和配置类代码。在此类中，我们将消息发布到指定的主题，并在我们在同一应用程序中定义的使用者类中侦听该消息。请注意，我们使用Spring自己的事件总线来承载作业，并且这些作业不会放在磁盘上。如果使用Spring Boot执行器正常终止了该应用程序，则这些作业将自动保留在磁盘上，以便在应用程序重新启动时可以重新排队。

在下一节中，我们将使用简单的Maven命令运行项目。

12.运行项目

既然完成了主类定义，我们就可以运行我们的项目。使用maven可以轻松运行应用程序，只需使用以下命令：

pom.xml

mvn spring-boot:run

一旦执行了以上命令，我们将看到一条消息已经发布，并且同一应用在事件处理程序中使用了该消息：

运行Spring Reactor应用程序

我们看到使用非阻塞模式下使用的CommandLineRunner方法启动应用程序时，事件已发布。事件发布后，事件处理程序将并行使用它。如果仔细研究使用者，您会注意到Spring在线程池中定义了四个线程来管理这些事件。这是Spring定义的用于并行管理事件的线程数的默认限制。

13.结论

在本课程中，我们研究了构建集成了Reactor项目的Spring Boot应用是多么容易和快捷。就像我们已经说过的那样，在您的应用程序中设计良好的反应堆模式可以具有每秒高达15,000,000（即六个零 ）事件的吞吐量。这表明该反应堆内部队列的执行效率如何。

在我们定义的小型应用程序中，我们演示了一种定义线程池执行程序的简单方法，该执行程序定义了四个线程，而使用者使用该线程池来并行管理事件。在依赖异步行为执行操作的应用程序中面临的最常见问题之一是，当有多个线程开始占用堆空间并在开始处理时创建对象时，它们很快就会耗尽内存。确保启动应用程序时，我们为应用程序分配良好的堆大小非常重要，这直接取决于为应用程序定义的线程池的大小。

反应式编程是最常见的编程风格之一，由于应用程序开始通过并行执行来利用CPU内核，因此这是一种正在兴起的编程风格，这是在应用程序级别使用硬件的好主意。 Reactor为JVM提供了完整的非阻塞编程基础，并且也可用于Groovy或Kotlin。由于Java本身不是反应性语言，因此它本身不支持协程。有多种JVM语言（例如Scala和Clojure）就本机性而言更好地支持反应模型，但是Java本身并没有做到这一点（至少直到版本9才如此）。