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支持两种类型的编程范式：
-
传统的基于注解的方式，如@Controller、@RequestMapping等沿用了Spring MVC的模式.
-
基于Java8的Lambda函数式编程模式
本文主要是使用基于注解的方式，今后另文补充基于函数式编程的范式。
基于Spring Boot 2+ Spring WebFlux + MongoDB的轻量级微博应用
以下展示如何搭建一个轻量级的微博应用，这个应用只包括一个domain类Tweet，使用基于MongoDB的在线MongoDB数据库mLab作为存储，并且使用异步的RESTful API提供基本的增删查改功能。
此外还会用到Spring Test组件，通过使用Maven的插件功能，实现对微服务应用的测试。
1. 新建项目
- 点击http://start.spring.io
- 选择2.x以上的Spring Boot版本
- 输入artifact的值，比如webflux-demo
- 选择Reactive Web和Reactive MongoDB依赖
- 点击Generate Project，生成并下载一个微服务框架到本地，并解压
- 使用IDE，比如eclipse，导入解压出来的项目文件
![]()
![]()
2. 注册mLab账户，并新建一个MongoDB数据库
MongoDB数据库是常用的文档类型数据库，广泛用于社交网站、电商等引用中。而mLab是一个在线MongoDB数据库平台，提供MongoDB的在线服务。这个应用使用到它。
- 前往https://mlab.com
- 根据要求注册账户
- 网站会有免费和收费的服务选择，选择AWS的免费MongoDB服务
- 服务选择完毕，平台会提供一个数据库镜像，可以点击数据库前往管理页面。
- 在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的数据库，数据被成功添加：
![]()
