Spring 5:使用 Spring Webflux 开发 Reactive 应用

Posted OSC开源社区

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spring 5:使用 Spring Webflux 开发 Reactive 应用相关的知识,希望对你有一定的参考价值。



OSC 协作翻译

原文:Spring 5 Reactive Web

链接:https://dzone.com/articles/spring-5-reactive-web-services

译者:Tocy, imqipan, chloe900, 无若, tnjin, sniff_hu, 总长


Spring 5 - Spring webflux 是一个新的非堵塞函数式 Reactive Web 框架,可以用来建立异步的,非阻塞,事件驱动的服务,并且扩展性非常好。


把阻塞(不可避免的)风格的代码迁移到函数式的非阻塞 Reactive 风格代码,需要把商业逻辑作为异步函数来调用。这可以参考 Java 8 的方法或者 lambda 表达式。由于线程是非阻塞的,处理能力能被最大化使用。


在发布这篇文章的时候,Spring 5 还处于一个里程碑版本中(5.0.0 M5)。



创建一个 Spring Boost 项目

可以通过 Spring initializer 创建一个Spring Boot项目。将如下的依赖添加到 pom.xml 中

<dependencies>

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter</artifactId>

    </dependency>

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-webflux</artifactId>

    </dependency>

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-test</artifactId>

        <scope>test</scope>

    </dependency>

</dependencies>


Spring-boot-starter-webflux 包中带了 spring-webflux, netty。其他的依赖需要自行添加。


建立一个简单的用户数据表和从 list 中获取到 user 数据的 DTO 类。这仅仅是一个虚拟的数据 bean,但是这可以实时从其它的数据源像 Rdbms,MongoDb,或者 RestClient 加载数据。由于 JDBC 天生不是响应式的,所以任何对数据库的调用都会阻塞这个线程。MongoDB 有一个响应式的客户端驱动。在测试响应式 Web 服务时的进一步渲染时,REST 风格的调用不会导致任何的阻塞。

public class User {

public User(){}


public User(Long id, String user) {

this.id = id;

this.user = user;

}


private Long id;

private String user;


public Long getId() { return id; }

public void setId(Long id) { this.id = id; }

public String getUser() { return user; }

public void setUser(String user) { this.user = user; }

}


@Repository

public class UserRepository {

private final List<User> users = Arrays.asList(new User(1L, "User1"), new User(2L, "User2"));


public Mono<User> getUserById(String id) {

return Mono.justOrEmpty(users.stream().filter(user -> {

return user.getId().equals(Long.valueOf(id));

}).findFirst().orElse(null));

}


public Flux<User> getUsers() {

return Flux.fromIterable(users);

}

}


Mono 和 Flux 是由目标反应器提供的响应类型。Springs 还提供其他的响应流的实现,例如 RXJava。


Mono 和 Flux 是 Reactive streams 的发布者实现。Mono 是 0 或者任意单个值的发布,Flux 是 0 到任意值的发布。他们和 RXJava 中的 Flowable 和 Observable 类似。他们代替流向这些订阅者发布信息。


GetUserById() 返回一个 Mono<User> 对象,这个对象不论何时都会返回 0~1 个用户对象,GetUsers() 返回一连串变动的用户对象,不论何时都包含 0~n 个用户对象。


相比命令式编程风格,我们并不返回可用前阻塞线程的 User/List<User> 对象,而只是返回一个流的引用,流可以在后面访问 User/List<User>。


创建带有处理 HTTP 请求函数的 Handler 类


@Service

public class UserHandler {

@Autowired

private UserRepository userRepository;


public Mono<ServerResponse> handleGetUsers(ServerRequest request) {

return ServerResponse.ok().body(userRepository.getUsers(), User.class);

}


public Mono<ServerResponse> handleGetUserById(ServerRequest request) {

return userRepository.getUserById(request.pathVariable("id"))

.flatMap(user -> ServerResponse.ok().body(Mono.just(user), User.class))

.switchIfEmpty(ServerResponse.notFound().build());

}

}

handler 类就像 Spring Web 中的 Service beans 一样,我们需要编写该服务的大部分业务功能。ServerResponse 就像 Spring Web 中的 ResponseEntity 类一样,我们可以在 ServerResponse 对象中打包 Response 的数据、状态码、头信息等。 ServerResponse 有很多有用的默认方法,如 notFound(), ok(), accepted(), created()等,可用于创建不同类型的反馈。


UserHandler 有不同的方法,都返回 Mono<ServerResponse>; UserRepository.getUsers() 返回Flux<User>; 和 ServerResponse.ok().body(UserRepository.getUsers(), User.class) 可将此 Flux <User> 转换为 Mono<ServerResponse>,这表明只要可用时均可发起 ServerResponse 的流。UserRepository.getUserById()返回一个Mono<User>,ServerResponse.ok().body(Mono.just(user), User.class) 将此 Mono<User> 转换为Mono<ServerResponse>,这说明随时都可以发起 ServerResponse 的流。


在给定的路径变量(pathVariable)中没有找到用户时,ServerResponse.notFound().build() 返回一个 Mono<ServerResponse>,表名是一个返回 404 服务响应的流。


在命令式编程风格中,数据接收前线程会一直阻塞,这样使得其线程在数据到来前无法运行。而响应式编程中,我们定义一个获取数据的流,然后定义一个在数据到来后的回调函数操作。这样不会使线程堵塞,在数据被返回时,可用线程就用于执行。


创建一个定义应用程序路由的路由类


import static org.springframework.web.reactive.function.server.RequestPredicates.GET;

import static org.springframework.web.reactive.function.server.RequestPredicates.accept;

import static org.springframework.web.reactive.function.server.RouterFunctions.route;

@Configuration

public class Routes {

private UserHandler userHandler;


    public Routes(UserHandler userHandler) {

this.userHandler = userHandler;

}


@Bean

public RouterFunction<?> routerFunction() {

return route(GET("/api/user").and(accept(MediaType.APPLICATION_JSON)), userHandler::handleGetUsers)

.and(route(GET("/api/user/{id}").and(accept(MediaType.APPLICATION_JSON)), userHandler::handleGetUserById));

}

}

RouterFunction就像Spring Web中的@RequestMapping类一样。 RouterFunction用于定义Spring5应用程序的路由。 


RouterFunctions帮助器类有一个有用的方法,类似路由,可用于定义路由并构建RouterFunction对象。 RequestPredicates有许多有用的方法,如GET,POST,path,queryParam,accept,headers,contentType等,来定义路由并构建RouterFunction。 每个路由映射到一个处理程序方法,当接收到适当的HttpRequest时,该方法必须被调用。


Spring5还支持定义应用程序处理程序映射的@RequestMapping类型的控制器。 我们可以编写如下所示的控制器方法,以在@RequestMapping样式中创建类似的API。

@GetMapping("/user") public Mono<ServerResponse> handleGetUsers() {}


控制器方法返回Mono<ServerResponse>。

RouterFunction为应用程序提供了DSL类型的路由功能。 到目前为止,Springs不支持混合这两种类型。


创建HttpServerConfig类,用于创建HttpServer类


import org.springframework.http.server.reactive.HttpHandler;

import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;

import reactor.ipc.netty.http.server.HttpServer;

@Configuration

public class HttpServerConfig {

@Autowired

private Environment environment;


@Bean

public HttpServer httpServer(RouterFunction<?> routerFunction) {

HttpHandler httpHandler = RouterFunctions.toHttpHandler(routerFunction);

ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);

HttpServer server = HttpServer.create("localhost", Integer.valueOf(environment.getProperty("server.port")));

server.newHandler(adapter);

return server;

}

}

这将使用应用程序属性中定义的端口创建一个 netty HttpServer。Spring 支持的其他服务器也跟 Tomcat 和 undertow 一样。由于 netty 是异步的,而且天生基于事件驱动,因此更适合响应式的应用程序。Tomcat 使用 Java NIO 来实现 servlet 规范。Netty 是 NIO 的一个实现,它针对异步、事件驱动的非阻塞 IO 应用程序进行了优化。


Tomcat 服务器也可以按照如下代码所示的用法使用:

Tomcat tomcatServer = new Tomcat();

tomcatServer.setHostname("localhost");

tomcatServer.setPort(Integer.valueOf(environment.getProperty("server.port")));

Context rootContext = tomcatServer.addContext("", System.getProperty("java.io.tmpdir"));

ServletHttpHandlerAdapter servlet = new ServletHttpHandlerAdapter(httpHandler);

Tomcat.addServlet(rootContext, "httpHandlerServlet", servlet);

rootContext.addServletMapping("/", "httpHandlerServlet");

tomcatServer.start();


创建用于启动应用的Spring启动主类

@SpringBootApplication

public class Spring5ReactiveApplication {

public static void main(String[] args) throws IOException {

SpringApplication.run(Spring5ReactiveApplication.class, args);

}

}


测试应用

你可以使用任意诸如Postman、CURL等的HTTP测试工具测试该应用。


Spring测试也支持为响应式服务编写集成测试的功能。

@RunWith(SpringRunner.class)

@SpringBootTest(webEnvironment = WebEnvironment.DEFINED_PORT)

public class UserTest {

@Autowired

private WebTestClient webTestClient;


@Test

public void test() throws IOException {

FluxExchangeResult<User> result = webTestClient.get().uri("/api/user").accept(MediaType.APPLICATION_JSON)

.exchange().returnResult(User.class);

assert result.getStatus().value() == 200;

List<User> users = result.getResponseBody().collectList().block();

assert users.size() == 2;

assert users.iterator().next().getUser().equals("User1");

}


@Test

public void test1() throws IOException {

User user = webTestClient.get().uri("/api/user/1")

.accept(MediaType.APPLICATION_JSON).exchange().returnResult(User.class).getResponseBody().blockFirst();

assert user.getId() == 1;

assert user.getUser().equals("User1");

}


@Test

public void test2() throws IOException {

webTestClient.get().uri("/api/user/10").accept(MediaType.APPLICATION_JSON).exchange().expectStatus()

.isNotFound();

}

}


WebTestClient 和 TestRestTemplate 类似, 他们都有调用 Spring 启动应用的 rest 方法,并能够验证响应结果。在 test 的配置中,Spring 测试创建了一个 TestRestTemplate 的 bean。这里面有一个 WebClient,就跟 Spring Web 中的 RestTemplate 类似。这可用于处理响应式和非阻塞的 rest 调用。


WebClient.create("http://localhost:9000").get().uri("/api/user/1")

        .accept(MediaType.APPLICATION_JSON).exchange().flatMap(resp -> resp.bodyToMono(User.class)).block();


exchange()返回Mono<ClientResponse>,它在Emits clientResponse可用时表示一个流。


    block()阻塞线程执行,直到Mono返回User/List<User>,因为这是我们需要数据来验证响应的测试用例。

Spring Web 因其易于开发/调试而是必要的。使用Spring5响应式或Spring Web命令式服务的决定必须根据用例明智地做出。在许多情况下,只有命令式的可能会很好,但是在高可扩展性是关键因素时,响应式非阻塞将更适合。





推荐阅读

点击“阅读原文”查看更多精彩内容


以上是关于Spring 5:使用 Spring Webflux 开发 Reactive 应用的主要内容,如果未能解决你的问题,请参考以下文章

Spring 5:使用 Spring Webflux 开发 Reactive 应用

使用 spring boot 2.5.5 和 Spring Cloud 2.0.4 创建 pubsubInputChannel 时出错

8 -- 深入使用Spring -- 2...5 Spring 3.0 新增的注解

如何在 Spring webflux 应用程序中使用 Spring WebSessionIdResolver 和 Spring Security 5?

从 Spring 4.2 升级到 Spring 5.0.5 时预见到的问题

使用 spring-security-oauth2 2.0.5.RELEASE 保护 Spring REST 服务