linux怎么查看本机内存大小
248
2023-05-21
如何用Spring WebFlux构建Reactive REST API
【51CTO.com快译】
在本文中，我们将讨论如何使用Spring WebFlux来构建响应式REST API。在正式讨论之前，让我们首先来看看系统的开发，传统REST在实现中遇到的问题，以及当前API的普遍需求。
下图简要地罗列了传统应用和现代应用系统的主要特点。如今的系统讲求的是：分布式应用、云原生、高可用性和可扩展性。因此，有效地利用系统现有的资源是至关重要的。
应用程序API需求的演变
那么传统的REST API请求处理又是如何工作的呢？
传统REST API模型
如上图所示，传统REST API会带来如下问题：
阻塞和同步 → 通常，请求线程会去等待各种阻塞的I/O直至结束之后，才能被释放，进而将响应返回给调用方。 每个请求的线程数 → Web容器会用到基于请求的线程（thread-per-request）模型。该模型限制了待处理的并发请求数量。也就是说，容器会对请求进行排队，进而最终影响到API的性能。 处理高并发用户的限制 → 正是由于Web容器使用了基于请求的线程模型，因此我们无法去处理那些高并发量的请求。 无法更好地利用系统资源 → 阻塞的I/O会造成线程处于空闲状态，进而导致Web容器无法接受更多的请求，我们也就无法有效地利用现有的系统资源。 没有背压（backpressure）支持 → 由于我们无法从客户端或服务器处施加背压，因此应用程序在负载量大时，无法维持正常运行。也就是说，倘若应用突然面临大量的请求，则服务器或客户端可能会由于中断，而无法访问到该应用。下面，让我们来看看响应式API的优势，以及如何使用响应式编程，来解决上述问题。
异步和无阻塞 → 响应式编程为编写异步和非阻塞应用程序提供了灵活性。 事件/消息驱动 → 系统能够为任何活动生成对应的事件或消息。例如，那些来自数据库的数据会被视为事件流。 支持背压 → 我们可以通过施加背压，来“优雅地”处理从一个系统到另一个系统的压力，从而避免了拒绝服务的出现。 可预测的应用响应时间 → 由于线程是异步且非阻塞的，因此我们可以预测负载下的应用响应时间。 更好地利用系统资源 → 同样由于线程是异步且非阻塞的，因此各种线程不会被I/O所占用，它们能够支持更多的用户请求。 基于负载的扩容方式 摆脱基于请求的线程 → 借助响应式API，并得益于异步且非阻塞的线程，我们可以摆脱基于请求的线程模型。在请求被产生后，模型会与服务器一起创建事件，并通过请求线程，去处理其他的请求。那么，响应式编程的具体流程是怎样的呢？如下图所示，一旦应用程序调用了从某个数据源获取数据的操作，那么就会立即返回一个线程，并且让来自该数据源的数据作为数据/事件流出现。在此，应用程序是订阅者（subscriber），数据源是发布者（publisher）。一旦数据流完成后，onComplete事件就会被触发。
数据流工作流程
如下图所示，如果发生了任何异常情况，发布者将会触发onError事件。
数据流工作流程
在某些情况下，例如：从数据库中删除一个条目，发布者只会立即触发onComplete/onError事件，而不会调用onNext事件，毕竟没有任何数据可以返回。
数据流工作流程
下面，我们进一步讨论：什么是背压，以及如何将背压应用于响应流。例如，我们有一个客户端应用正在向另一个服务请求数据。该服务能够以1000 TPS（吞吐量）的速率发布事件，而客户端应用只能以200 TPS的速率处理事件。
那么在这种情况下，客户端应用程序需要通过缓冲数据来进行处理。而在随后的调用中，客户端应用程序可能会缓冲更多的数据，以致最终耗尽内存。显然，这对于那些依赖该客户端应用的其他程序，会造成级联效应。为了避免此类情况，客户端应用可以要求服务在事件的末尾进行缓冲，并以客户端应用的速率去推送各种事件。这就是所谓的背压，具体流程请见下图。
背压示例
下面，我们将介绍响应流的规范（请参见--https://www.reactive-streams.org/），以及一个实现案例--Project Reactor（请参见--https://projectreactor.io/）。通常，响应流的规范定义了如下接口类型：
发布者（Publisher） → 发布者是那些具有无限数量顺序元素的提供者。它可以按照订阅者的要求进行发布。其Java代码如下所示：复制public interface Publisher<T> {
public void subscribe(Subscriber<? super T> s);
}
1.2.3.
复制public interface Subscriber<T> {
public void onSubscribe(Subscription s);
public void onNext(T t);
public void onError(Throwable t);
public void onComplete();
}
1.2.3.4.5.6.
复制public interface Subscription {
public void request(long n);
public void cancel();
}
1.2.3.4.
下面是响应流规范的类图：
响应流规范
其实，响应流规范具有许多种实现方式，上述Project Reactor只是其中的一种。Reactor可以完全实现无阻塞、且有效的请求管理。它能够提供两个响应式和可组合的API，即：Flux [N]（请参见--https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html）和Mono [0|1]（请参见--https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html）。它们广泛地实现了响应式扩展（Reactive Extensions）。Reactor为HTTP（包括Websocket）提供了非阻塞的背压式网络引擎、TCP和UDP。它也非常适合于微服务的架构。
Flux→ 它是发布者带有各种rx运算符的响应流（Reactive Streams），它会发出0到N个元素，然后输出成功、或带有某个错误的完成结果。其流程图如下所示：图片来源：https://projectreactor.io
Mono → 它也是发布者具有各种基本rx运算符的响应流，能够通过发出0到1个元素，输出成功、或带有某个错误的完成结果。其流程图如下所示：图片来源：https://projectreactor.io
由于Reactor的实施往往涉及到Spring 5.x，因此，我们可以使用带有Spring servlet栈的命令式编程，来构建REST API。下图展示了Spring如何支持响应式和servlet栈的实现。
图片来源：spring.io
下面是一个公布了响应式REST API的应用。在该应用中，我们使用到了：
带有WebFlux的Spring Boot 具有响应式支持的Spring数据 Cassandra数据库下图是该应用的整体架构：
下面是build.gradle文件的Groovy代码，它包含了与Spring WebFlux协同使用的各种依赖项。
复制plugins {
id org.springframework.boot version 2.2.6.RELEASE
id io.spring.dependency-management version 1.0.9.RELEASE
id java
}
group = org.smarttechie
version = 0.0.1-SNAPSHOT
sourceCompatibility = 1.8
repositories {
mavenCentral()
}
dependencies {
implementation org.springframework.boot:spring-boot-starter-data-cassandra-reactive
implementation org.springframework.boot:spring-boot-starter-webflux
testImplementation(org.springframework.boot:spring-boot-starter-test) {
exclude group: org.junit.vintage, module: junit-vintage-engine
}
testImplementation io.projectreactor:reactor-test
}
test {
useJUnitPlatform()
}
1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.
在此应用程序中，我公布了如下API。您可以通过GitHub的相关链接--https://github.com/2013techsmarts/Spring-Reactive-Examples，下载源代码。
在构建响应式API时，我们可以使用功能性样式编程模型来构建API，而无需使用RestController。当然，您需要具有如下的router和handler组件：
Router：
复制package org.smarttechie.router;
import org.smarttechie.handler.ProductHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import static org.springframework.web.reactive.function.server.RequestPredicates.*;
@Configuration
public class ProductRouter {
/**
* The router configuration for the product handler.
* @param productHandler
* @return
*/
@Bean
public RouterFunction<ServerResponse> productsRoute(ProductHandler productHandler){
return RouterFunctions.route(GET("/products").and(accept(MediaType.APPLICATION_JSON)) ,productHandler::getAllProducts).andRoute(POST("/product").and(accept(MediaType.APPLICATION_JSON)),productHandler::createProduct).andRoute(DELETE("/product/{id}").and(accept(MediaType.APPLICATION_JSON)) ,productHandler::deleteProduct).andRoute(PUT("/product/{id}").and(accept(MediaType.APPLICATION_JSON)),productHandler::updateProduct);
}
}
1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.
Handler：
复制package org.smarttechie.handler;
import org.smarttechie.model.Product;
import org.smarttechie.service.ProductService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;
import static org.springframework.web.reactive.function.BodyInserters.fromObject;
@Component
public class ProductHandler {
@Autowired
private ProductService productService;
static Mono<ServerResponse> notFound = ServerResponse.notFound().build();
/**
* The handler to get all the available products.
* @param serverRequest
* @return - all the products info as part of ServerResponse
*/
public Mono<ServerResponse> getAllProducts(ServerRequest serverRequest) {
return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(productService.getAllProducts(), Product.class);
}
/**
* The handler tocreate a product
* @param serverRequest
* @return - return the created product as part of ServerResponse
*/
public Mono<ServerResponse> createProduct(ServerRequest serverRequest) {
Mono<Product> productToSave = serverRequest.bodyToMono(Product.class);
return productToSave.flatMap(product ->
ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(productService.save(product), Product.class));
}
/**
* The handler todelete a product based on the product id.
* @param serverRequest
* @return - return the deleted product as part of ServerResponse
*/
public Mono<ServerResponse> deleteProduct(ServerRequest serverRequest) {
String id = serverRequest.pathVariable("id");
Mono<Void> deleteItem = productService.deleteProduct(Integer.parseInt(id));
return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(deleteItem, Void.class);
}
/**
* The handler toupdate a product.
* @param serverRequest
* @return - The updated product as part of ServerResponse
*/
public Mono<ServerResponse> updateProduct(ServerRequest serverRequest) {
return productService.update(serverRequest.bodyToMono(Product.class)).flatMap(product ->ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(fromObject(product))) .switchIfEmpty(notFound);
}
}
1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.48.49.50.51.52.53.54.55.
至止，我们已经对如何公布响应式REST API有所了解。针对上述实现，我们使用了Gatling（译者注：是一款功能强大的负载测试工具），在响应式API和非响应式API（使用Spring RestController构建非响应式API）上，进行了简单的基准化测试。其结果比较如下图所示。具体的Gatling负载测试脚本，请参考GitHub上的链接：https://github.com/2013techsmarts/Spring-Reactive-Examples。
负载测试结果比较
原标题:Build Reactive REST APIs With Spring WebFlux ,作者：Siva Prasad Rao Janapati
【51CTO译稿，合作站点转载请注明原文译者和出处为51CTO.com】
版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。
相关文章
发表评论
暂时没有评论，来抢沙发吧~