实战基于Spring Boot 2的WebFlux和mLab搭建反应式Web

Spring Framework 5带来了新的Reactive Stack非阻塞式Web框架：Spring WebFlux。作为与Spring MVC并行使用的Web框架，Spring WebFlux依赖了反应式流适配器(Reactive Streams Adapter)，在Netty和Servlet3.1的容器下，可以提供非阻塞式的Web服务，充分发挥下一代多核处理器的优势，支撑海量的并发访问。

以上是官网的介绍，事实上在基于Spring Boot 2强大的微服务架构帮助下，WebFlux和Spring MVC一起，成为Java应用开发的两大选择，可以让我们迅速地搭建起反应式的Web应用。本文拟通过模拟一个简单的微博应用，实战通过Spring Boot 2+ Spring WebFlux + MongoDB 开发一个Web应用。

Spring WebFlux及其编程范式

Spring WebFlux通过核心库Reactor提供反应式支持，Reactor实现了Reactive Streams，后者是一个带非阻塞式背压的异步流处理器。

Reactor包含两个重要的成员变量Flux和Mono，它们都实现了Reactive Streams提供的Publisher接口. Flux 是一个代表了0..N元素的流，Mono是代表了一个0..1元素的流。虽然WebFlux使用Reactor作为它的核心依赖，它在应用层面，它也同时支持RxJava。

Spring WebFlux支持两种类型的编程范式：

1. 传统的基于注解的方式，如@Controller、@RequestMapping等沿用了Spring MVC的模式.

2. 基于Java8的Lambda函数式编程模式

本文主要是使用基于注解的方式，今后另文补充基于函数式编程的范式。

基于Spring Boot 2+ Spring WebFlux + MongoDB的轻量级微博应用

以下展示如何搭建一个轻量级的微博应用，这个应用只包括一个domain类Tweet，使用基于MongoDB的在线MongoDB数据库mLab作为存储，并且使用异步的RESTful API提供基本的增删查改功能。

此外还会用到Spring Test组件，通过使用Maven的插件功能，实现对微服务应用的测试。

1. 新建项目

1. 点击http://start.spring.io
2. 选择2.x以上的Spring Boot版本
3. 输入artifact的值，比如webflux-demo
4. 选择Reactive Web和Reactive MongoDB依赖
5. 点击Generate Project，生成并下载一个微服务框架到本地，并解压
6. 使用IDE，比如eclipse，导入解压出来的项目文件

2. 注册mLab账户，并新建一个MongoDB数据库

MongoDB数据库是常用的文档类型数据库，广泛用于社交网站、电商等引用中。而mLab是一个在线MongoDB数据库平台，提供MongoDB的在线服务。这个应用使用到它。

1. 前往https://mlab.com
2. 根据要求注册账户
3. 网站会有免费和收费的服务选择，选择AWS的免费MongoDB服务
4. 服务选择完毕，平台会提供一个数据库镜像，可以点击数据库前往管理页面。
5. 在User标签下，新建数据库的登录名和密码。

完成以上步骤，数据库就可以开始使用了。你会看到如下图所示的页面：

3. 在项目中配置MongoDB数据库

前往IDE中的项目资源文件夹，找到application.properties。添加你在mLad的MongoDB URI

spring.data.mongodb.uri=mongodb://username:password@ds063439.mlab.com:63439/springdb 

在应用启动的时候，Springboot会自动读取该配置文件。

4. 编写应用各模块

WebFlux可以认为是基于Spring的Web开发的一个新的模式或选择，因此它既有Spring MVC有的模块如Domain、Controller、Service，也有新增的如Handler、Router等。下面分别编写各模块。

4.1 Domain包

Domain包只包括一个domain类Tweet.java，因为使用了文档数据库，因此使用@Document注解修饰类，并且使用@Id修饰成员变量id。@NotBlank、@NotNull和@Size限定了成员变量的值的范围。代码如下：

@Document(collection = "tweets") public class Tweet { @Id private String id; @NotBlank @Size(max = 140) private String text; @NotNull private Date createAt = new Date(); public Tweet() { } //省略Tweet的getter和setter方法 }

4.2 Repository

Repository接口是DAO，继承了ReactiveMongoRepository接口用于连接MongoDB数据库做数据持久化，

 1 package com.example.webfluxdemo.repository;  2  3 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;  4 import org.springframework.stereotype.Repository;  5  6 import com.example.webfluxdemo.model.Tweet;  7  8 @Repository  9 public interface TweetRepository extends ReactiveMongoRepository<Tweet, String> { 10 11 }

其中父接口ReactiveMongoRepository的源码如下：

public interface ReactiveMongoRepository<T, ID> extends ReactiveSortingRepository<T, ID>, ReactiveQueryByExampleExecutor<T> { <S extends T> Mono<S> insert(S entity); <S extends T> Flux<S> insert(Iterable<S> entities); <S extends T> Flux<S> insert(Publisher<S> entities); <S extends T> Flux<S> findAll(Example<S> example); <S extends T> Flux<S> findAll(Example<S> example, Sort sort); }

通过查看源码可知，父接口ReactiveMongoRepository包含对MongoDB数据库基本的增删改查方法。在运行时，Spring Boot会自动实现一个SimpleReactiveMongoRepository类，用于执行增删改查方法。这样极大地节省了程序员持久化的精力，可以专注于业务开发。

4.3 Controller

Controller是WebFlux的核心类，该类定义了增删查改对应的方法，代码如下：

@RestController public class TweetController { @Autowired private TweetRepository tweetRepository; //通过接受Get请求，返回Flux类型的Tweet对象流 @GetMapping("/tweets") public Flux<Tweet> getAllTweets(){ return tweetRepository.findAll(); } //通过接受POST请求，新增一个Tweet对象 @PostMapping("/tweets") public Mono<Tweet> createTweets(@Valid @RequestBody Tweet tweet){ return tweetRepository.save(tweet); } //通过id查找Tweet @GetMapping("/tweets/{id}") public Mono<ResponseEntity<Tweet>> getTweetById(@PathVariable(value = "id") String tweetId) { return tweetRepository.findById(tweetId) .map(savedTweet -> ResponseEntity.ok(savedTweet)) .defaultIfEmpty(ResponseEntity.notFound().build()); } //通过id更新Tweet，使用到SpringMVC的相关注解 @PutMapping("/tweets/{id}") public Mono<ResponseEntity<Tweet>> updateTweet(@PathVariable(value = "id") String tweetId, @Valid @RequestBody Tweet tweet) { return tweetRepository.findById(tweetId) .flatMap(existingTweet -> { existingTweet.setText(tweet.getText()); return tweetRepository.save(existingTweet); }) .map(updatedTweet -> new ResponseEntity<>(updatedTweet, HttpStatus.OK)) .defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND)); } //通过id删除tweet @DeleteMapping("/tweets/{id}") public Mono<ResponseEntity<Void>> deleteTweet(@PathVariable(value = "id") String tweetId) { return tweetRepository.findById(tweetId) .flatMap(existingTweet -> tweetRepository.delete(existingTweet) .then(Mono.just(new ResponseEntity<Void>(HttpStatus.OK))) ) .defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND)); } // 基于反应式流发送微博至客户端 @GetMapping(value = "/stream/tweets", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux<Tweet> streamAllTweets() { return tweetRepository.findAll(); } } 

Controller是FluxWeb编程的核心，与SpringMVC不同，所有的处理方法返回的都是Flux或Mono对象。

Flux 和 Mono 是 Reactor 中的两个基本概念。Flux 表示的是包含 0 到 N 个元素的异步序列。在该序列中可以包含三种不同类型的消息通知：正常的包含元素的消息、序列结束的消息和序列出错的消息。当消息通知产生时，订阅者中对应的方法 onNext(), onComplete()和 onError()会被调用。

Mono 表示的是包含 0 或者 1 个元素的异步序列。该序列中同样可以包含与 Flux 相同的三种类型的消息通知。

Flux 和 Mono 之间可以进行转换。对一个 Flux 序列进行计数操作，得到的结果是一个 Mono<Long>对象。把两个 Mono 序列合并在一起，得到的是一个 Flux 对象。

Controller使用Flux或Mono作为对象，返回给不同的请求。反应式编码主要在最后一个方法：

// 基于反应式流发送微博至客户端 @GetMapping(value = "/stream/tweets", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux<Tweet> streamAllTweets() { return tweetRepository.findAll(); } 

这个方法和getAllTweet方法一样，会返回一个JSON流到客户端，区别在于streamAllTweets以Server-send-event的方式返回一个Json流到浏览器，这种流可以被浏览器识别和使用。这里涉及到服务器推送事件(Server-Send Event)

服务器推送事件（Server-Sent Events，SSE）允许服务器端不断地推送数据到客户端。相对于 WebSocket 而言，服务器推送事件只支持服务器端到客户端的单向数据传递。虽然功能较弱，但优势在于 SSE 在已有的 HTTP 协议上使用简单易懂的文本格式（如JSON）来表示传输的数据。

作为 W3C 的推荐规范，SSE 在浏览器端的支持也比较广泛，除了 IE 之外的其他浏览器都提供了支持。在 IE 上也可以使用 polyfill 库来提供支持。在服务器端来说，SSE 是一个不断产生新数据的流，非常适合于用反应式流来表示。在 WebFlux 中创建 SSE 的服务器端是非常简单的。只需要返回的对象的类型是 Flux<ServerSentEvent>，就会被自动按照 SSE 规范要求的格式来发送响应。

使用WebTestClient测试应用

WebTestClient是Spring 5提供的一个异步反应式Http客户端，可以用于测试反应式的RestFul微服务应用。在IDE的测试文件夹中，可以找到测试类，编写代码如下：

public class WebfluxDemoApplicationTests { @Autowired private WebTestClient webTestClient; @Autowired TweetRepository tweetRepository; @Test public void testCreateTweet() { Tweet tweet = new Tweet("这是一条测试微博"); webTestClient.post().uri("/tweets") .contentType(MediaType.APPLICATION_JSON_UTF8) .accept(MediaType.APPLICATION_JSON_UTF8) .body(Mono.just(tweet), Tweet.class) .exchange() .expectStatus().isOk() .expectHeader().contentType(MediaType.APPLICATION_JSON_UTF8) .expectBody() .jsonPath("$.id").isNotEmpty() .jsonPath("$.text").isEqualTo("这是一条测试微博"); } } 

在测试类中通过控制反转注入WebTestClient和DAO的对象，调用WebTestClient方法进行测试，使用mvn test命令，测试所有的测试类。结果如下：

查看mLab的数据库，数据被成功添加：