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

创建一个Spring Boost项目

可以通过 Spring initializer 创建一个Spring Boot项目。将如下的依赖添加到 pox.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 类似。他们代替流向这些订阅者发布信息。

创建带有处理 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 的流。

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

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命令式服务的决定必须根据用例明智地做出。在许多情况下，只有命令式的可能会很好，但是在高可扩展性是关键因素时，响应式非阻塞将更适合。