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

发布日期:2019-04-28

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包含两个重要的成员变量FluxMono,它们都实现了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 {@Idprivate String id@NotBlank@Size(max = 140)private String text@NotNullprivate 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的核心类,该类定义了增删查改对应的方法,代码如下:

@RestControllerpublic class TweetController {@Autowiredprivate 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 {@Autowiredprivate WebTestClient webTestClient@AutowiredTweetRepository tweetRepository@Testpublic 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的数据库,数据被成功添加: