聊聊 Spring Boot 2.0 的 WebFlux## 前言
对照下 Spring Web MVC ，Spring Web MVC 是基于 Servlet API 和 Servlet 容器设计的。那么 Spring WebFlux 确定不是基于前面二者，它基于 Reactive Streams API 和 Servlet 3.1+ 容器设计。java
那 Reactive Streams API 是什么？
先理解 Stream 流是什么？流是序列，是生产者生产，一个或多个消费者消费的元素序列。这种具体的设计模式成为发布订阅模式。常见的流处理机制是 pull / push 模式。背压是一种经常使用策略，使得发布者拥有无限制的缓冲区存储 item，用于确保发布者发布 item 太快时，不会去压制订阅者。
Reactive Streams （响应式流）是提供处理非阻塞背压异步流的一种标准。主要针对的场景是运行时环境（包括 JVM 和 JS）和网络。一样，JDK 9 java.util.concurrent 包提供了两个主要的 API 来处理响应流：
- Flow
- SubmissionPublisherreact
为啥只能运行在 Servlet 3.1+ 容器？
你们知道，3.1 规范其中一个新特性是异步处理支持。
异步处理支持：Servlet 线程不需一直阻塞，即不须要到业务处理完毕再输出响应，而后结束 Servlet线程。异步处理的做用是在接收到请求以后，Servlet 线程能够将耗时的操做委派给另外一个线程来完成，在不生成响应的状况下返回至容器。主要应用场景是针对业务处理较耗时的状况，能够减小服务器资源的占用，而且提升并发处理速度。
因此 WebFlux 支持的容器有 Tomcat、Jetty（Non-Blocking IO API） ，也能够像 Netty 和 Undertow 的自己就支持异步容器。在容器中 Spring WebFlux 会将输入流适配成 Mono 或者 Flux 格式进行统一处理。git
Spring WebFlux 是什么
先看这张图，上面咱们了解了容器、响应流。这里介绍下 Spring WebFlux 是什么？ Spring WebFlux 是 Spring 5 的一个新模块，包含了响应式 HTTP 和 WebSocket 的支持，另外在上层服务端支持两种不一样的编程模型：
- 基于 Spring MVC 注解 @Controller 等
- 基于 Functional 函数式路由github
下面是两个实现小案例，首先在 pom.xml 加入对应的依赖:web
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
基于 Spring MVC 注解 RESTful API
官方案例很简单，以下：spring
@RestController
public class PersonController {
private final PersonRepository repository;
public PersonController(PersonRepository repository) {
this.repository = repository;
}
@PostMapping("/person")
Mono<Void> create(@RequestBody Publisher<Person> personStream) {
return this.repository.save(personStream).then();
}
@GetMapping("/person")
Flux<Person> list() {
return this.repository.findAll();
}
@GetMapping("/person/{id}")
Mono<Person> findById(@PathVariable String id) {
return this.repository.findOne(id);
}
}
可是 PersonRepository 这种 Spring Data Reactive Repositories 不支持 MySQL，进一步也不支持 MySQL 事务。因此用了 Reactivey 原来的 spring 事务管理就很差用了。jdbc jpa 的事务是基于阻塞 IO 模型的，若是 Spring Data Reactive 没有升级 IO 模型去支持 JDBC，生产上的应用只能使用不强依赖事务的。也可使用透明的事务管理，即每次操做的时候以回调形式去传递数据库链接 connection。数据库
Spring Data Reactive Repositories 目前支持 Mongo、Cassandra、Redis、Couchbase 。编程
那提到的“用了 Reactivey 原来的 spring 事务管理就很差用了”，您可否再详细介绍一下？另外，应用有强依赖事务，有没有对应的解决方案？
咱们先看看这张图。Spring Boot 2.0 这里有两条不一样的线分别是：
Spring Web MVC -> Spring Data
Spring WebFlux -> Spring Data Reactive设计模式
因此这里问题的答案是，若是使用 Spring Data Reactive ，原来的 Spring 针对 Spring Data （JDBC等）的事务管理确定不起做用了。由于原来的 Spring 事务管理（Spring Data JPA）都是基于 ThreadLocal 传递事务的，其本质是基于 阻塞 IO 模型，不是异步的。但 Reactive 是要求异步的，不一样线程里面 ThreadLocal 确定取不到值了。天然，咱们得想一想如何在使用 Reactive 编程是作到事务，有一种方式是 回调 方式，一直传递 conn ：
newTransaction(conn ->{})api
由于每次操做数据库也是异步的，因此 connection 在 Reactive 编程中没法靠 ThreadLocal 传递了，只能放在参数上面传递。虽然会有必定的代码侵入行。进一步，也能够 kotlin 协程，去作到透明的事务管理，即把 conn 放到 协程的局部变量中去。
那 Spring Data Reactive Repositories 不支持 MySQL，进一步也不支持 MySQL 事务，怎么办？
答案是，这个问题其实和第一个问题也相关。 为啥不支持 MySQL，即 JDBC 不支持。你们能够看到 JDBC 是所属 Spring Data 的。因此能够等待 Spring Data Reactive Repositories 升级 IO 模型，去支持 MySQL。也能够和上面也讲到了，如何使用 Reactive 编程支持事务。
若是应用只能使用不强依赖数据事务，依旧使用 MySQL ，可使用下面的实现，代码以下：
@RestController
@RequestMapping(value = "/city")
public class CityRestController {
@Autowired
private CityService cityService;
@RequestMapping(value = "/{id}", method = RequestMethod.GET)
public Mono<City> findOneCity(@PathVariable("id") Long id) {
return Mono.create(cityMonoSink -> cityMonoSink.success(cityService.findCityById(id)));
}
@RequestMapping(method = RequestMethod.GET)
public Flux<City> findAllCity() {
return Flux.create(cityFluxSink -> {
cityService.findAllCity().forEach(city -> {
cityFluxSink.next(city);
});
cityFluxSink.complete();
});
}
@RequestMapping(method = RequestMethod.POST)
public Mono<Long> createCity(@RequestBody City city) {
return Mono.create(cityMonoSink -> cityMonoSink.success(cityService.saveCity(city)));
}
@RequestMapping(method = RequestMethod.PUT)
public Mono<Long> modifyCity(@RequestBody City city) {
return Mono.create(cityMonoSink -> cityMonoSink.success(cityService.updateCity(city)));
}
@RequestMapping(value = "/{id}", method = RequestMethod.DELETE)
public Mono<Long> modifyCity(@PathVariable("id") Long id) {
return Mono.create(cityMonoSink -> cityMonoSink.success(cityService.deleteCity(id)));
}
}
findAllCity 方法中，利用 Flux.create 方法对响应进行建立封装成 Flux 数据。而且使用 lambda 写数据流的处理函数会十分的方便。
Service 层依旧是之前那套逻辑，业务服务层接口以下：
public interface CityService {
/**
* 获取城市信息列表
*
* @return
*/
List<City> findAllCity();
/**
* 根据城市 ID,查询城市信息
*
* @param id
* @return
*/
City findCityById(Long id);
/**
* 新增城市信息
*
* @param city
* @return
*/
Long saveCity(City city);
/**
* 更新城市信息
*
* @param city
* @return
*/
Long updateCity(City city);
/**
* 根据城市 ID,删除城市信息
*
* @param id
* @return
*/
Long deleteCity(Long id);
}
具体案例在个人 Github：https://github.com/JeffLi1993/springboot-learning-example
基于 Functional 函数式路由实现 RESTful API
建立一个 Route 类来定义 RESTful HTTP 路由
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 CityService cityService;
public Routes(CityService cityService) {
this.cityService = cityService;
}
@Bean
public RouterFunction<?> routerFunction() {
return route(
GET("/api/city").and(accept(MediaType.APPLICATION_JSON)), cityService:: findAllCity).and(route(
GET("/api/user/{id}").and(accept(MediaType.APPLICATION_JSON)), cityService:: findCityById)
);
}
}
RoouterFunction 相似 Spring Web MVC 的 @RequestMapping ，用来定义路由信息，每一个路由会映射到一个处理方法，当接受 HTTP 请求时候会调用该处理方法。
建立 HttpServerConfig 自定义 Http Server，这里建立一个 Netty HTTP 服务器：
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 来运行 Reactive 应用，由于 Netty 是基于异步和事件驱动的。