## WebFlux 简介
WebFlux 模块的名称是 spring-webflux,名称中的 Flux 来源于 Reactor 中的类 Flux。该模块中包含了对反应式 HTTP、服务器推送事件和 WebSocket 的客户端和服务器端的支持。对于开发人员来说,比较重要的是服务器端的开发,这也是本文的重点。在服务器端,WebFlux 支持两种不同的编程模型:第一种是 Spring MVC 中使用的基于 Java 注解的方式;第二种是基于 Java 8 的 lambda 表达式的函数式编程模型。这两种编程模型只是在代码编写方式上存在不同。它们运行在同样的反应式底层架构之上,因此在运行时是相同的。WebFlux 需要底层提供运行时的支持,WebFlux 可以运行在支持 Servlet 3.1 非阻塞 IO API 的 Servlet 容器上,或是其他异步运行时环境,如 Netty 和 Undertow。
## 什么是反应式编程(Reactor )?
简而言之,反应式编程是关于非阻塞应用程序,它们是异步的和事件驱动的,并且需要少量线程来垂直(即在JVM中)而不是水平(即通过集群)扩展。
反应性应用程序的一个关键方面是背压的概念,它是确保生产者不会压倒消费者的一种机制。例如,在反应性组件从数据库扩展到HTTP响应的管道中,当HTTP连接太慢时,数据存储库也可能变慢或完全停止,直到释放网络容量。
反应式编程还导致从命令式逻辑到声明式异步逻辑的重大转变。与使用CompletableFutureJava 8通过lambda表达式编写后续动作相比,它相当于编写阻塞代码 。

## Spring MVC还是WebFlux?
一个自然的问题要问,但建立了一个不合理的二分法。实际上,两者共同努力扩大了可用选项的范围。两者的设计旨在实现彼此的连续性和一致性,它们可以并行使用,并且来自双方的反馈对双方都有利。下图显示了两者之间的关系,它们的共同点以及各自的独特支持:

我们建议您考虑以下几点:
如果您有运行正常的Spring MVC应用程序,则无需更改。命令式编程是编写,理解和调试代码的最简单方法。您有最大的库选择空间,因为从历史上看,大多数库都处于阻塞状态。
如果您已经在购买无阻塞Web堆栈,那么Spring WebFlux在此空间中提供的执行模型优势与其他模型相同,并且还提供服务器选择(Netty,Tomcat,Jetty,Undertow和Servlet 3.1+容器),选择编程模型(带注释的控制器和功能性Web端点),以及选择反应式库(Reactor,RxJava或其他)。
如果您对与Java 8 lambda或Kotlin一起使用的轻量级功能性Web框架感兴趣,则可以使用Spring WebFlux功能性Web端点。对于要求较低复杂性的较小应用程序或微服务(可以受益于更高的透明度和控制)而言,这也是一个不错的选择。
在微服务架构中,您可以混合使用带有Spring MVC或Spring WebFlux控制器或带有Spring WebFlux功能端点的应用程序。在两个框架中都支持相同的基于注释的编程模型,这使得重用知识变得更加容易,同时还为正确的工作选择了正确的工具。
评估应用程序的一种简单方法是检查其依赖关系。如果您要使用阻塞持久性API(JPA,JDBC)或网络API,则Spring MVC至少是常见体系结构的最佳选择。使用Reactor和RxJava在单独的线程上执行阻塞调用在技术上是可行的,但是您不会充分利用非阻塞Web堆栈。
如果您的Spring MVC应用程序具有对远程服务的调用,请尝试使用active WebClient。您可以直接从Spring MVC控制器方法返回反应类型(Reactor,RxJava 或其他)。每个呼叫的等待时间或呼叫之间的相互依赖性越大,好处就越明显。Spring MVC控制器也可以调用其他反应式组件。
如果您有庞大的团队,请牢记向无阻塞,功能性和声明性编程的过渡过程中的学习曲线陡峭。在没有完全切换的情况下启动的实际方法是使用电抗器WebClient。除此之外,从小处着手并衡量收益。我们希望对于广泛的应用而言,这种转变是不必要的。如果不确定要寻找什么好处,请先了解无阻塞I / O的工作原理(例如,单线程Node.js上的并发性)及其影响。
## WebFlux支持2种不同的编程模型:
* 基于注释的@Controller和MVC还支持其他注释
* 功能性Java 8 Lambda样式的路由和处理
两种编程模型都是在同一反应式基础上执行的,该反应式基础将非阻塞HTTP运行时适应于Reactive Streams API。下图显示了服务器端堆栈,其中包括spring-webmvc模块左侧的传统的基于Servlet的Spring MVC,以及模块右侧的反应式堆栈spring-webflux。

WebFlux可以在支持Servlet 3.1 Non-Blocking IO API的Servlet容器以及其他异步运行时(例如Netty和Undertow)上运行。每个运行时都适合于响应式, ServerHttpRequest并且ServerHttpResponse将请求和响应的主体暴露为Flux<DataBuffer>,而不是 InputStream和OutputStream,而具有响应式背压。REST风格的JSON和XML序列化和反序列化作为顶部受支持Flux<Object>,HTML视图呈现和Server-Sent事件也受支持。
## 关于Mono和Flux
Mono和Flux都是Publisher(发布者)。
其实,对于大部分业务开发人员来说,当编写反应式代码时,我们通常只会接触到 Publisher 这个接口,对应到 Reactor 便是 Mono 和 Flux。对于 Subscriber 和 Subcription 这两个接口,Reactor 必然也有相应的实现。但是,这些都是 Web Flux 和 Spring Data Reactive 这样的框架用到的。如果不开发中间件,通常开发人员是不会接触到的。
比如,在 Web Flux,你的方法只需返回 Mono 或 Flux 即可。你的代码基本也只和 Mono 或 Flux 打交道。而 Web Flux 则会实现 Subscriber ,onNext 时将业务开发人员编写的 Mono 或 Flux 转换为 HTTP Response 返回给客户端。
# 实战
我们在来看一下下图,可以看到,在目前的Spring WebFlux还没有支持类似
Mysql这样的关系型数据库,本文以MongoDb数据库为例

pom
```
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
```
整合mongodb
* 配置文件
```
spring.data.mongodb.uri=mongodb://localhost:27017/webflux
```
* 启动类添加注解`@EnableReactiveMongoRepositories`
```
@SpringBootApplication
@EnableReactiveMongoRepositories
public class SpringbootWebfluxApplication {
public static void main(String[] args) {
SpringApplication.run(SpringbootWebfluxApplication.class, args);
}
}
```
#实现ReactiveMongoRepositor
```
public interface UserDao extends ReactiveMongoRepository<User,String> {
}
```
### 基于注释的@Controller和MVC还支持其他注释
编写 UserController
```
@RestController
@RequestMapping("/user")
public class UserController {
@Autowired
private UserDao userDao;
@GetMapping("/")
public Flux<User> GetAll() {
return userDao.findAll();
}
@GetMapping(value = "/stream/all", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<User> StreamGetAll() {
return userDao.findAll();
}
@GetMapping("/{id}")
public Mono<ResponseEntity<User>> GetId( @PathVariable String id) {
return userDao.findById(id).map(u -> new ResponseEntity<User>(u, HttpStatus.OK))
.defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
}
@PostMapping(value = "/")
public Mono<User> createUser(@Valid@RequestBody User user) {
user.setId(null);
user.setCreatedDate(LocalDateTime.now());
return userDao.save(user);
}
@PutMapping(value = "/{id}")
public Mono<ResponseEntity<User>> updateUser(@PathVariable String id, @Valid@RequestBody User user) {
return userDao.findById(id).flatMap(u -> {
u.setAge(user.getAge());
u.setName(user.getName());
u.setCreatedDate(LocalDateTime.now());
return userDao.save(u);
}
).map(u -> new ResponseEntity<User>(u, HttpStatus.OK))
.defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
}
@DeleteMapping(value = "/{id}")
public Mono<ResponseEntity<Void>> DeleteUser(@PathVariable String id) {
return userDao.findById(id)
.flatMap(u -> userDao.deleteById(id)
.then(Mono.just(new ResponseEntity<Void>(HttpStatus.OK))))
.defaultIfEmpty(new ResponseEntity<Void>(HttpStatus.NOT_FOUND));
}
}
```
### 功能性Java 8 Lambda样式的路由和处理
## 编写Handle
```
@Component
public class UserHandle {
@Autowired
private UserDao userDao;
public Mono<ServerResponse> GetAll (ServerRequest request){
return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
.body(userDao.findAll(), User.class);
}
public Mono<ServerResponse> StreamGetAll (ServerRequest request){
return ServerResponse.ok().contentType(MediaType.valueOf(MediaType.TEXT_EVENT_STREAM_VALUE))
.body(userDao.findAll(), User.class);
}
public Mono<ServerResponse> createUser (ServerRequest request){
Mono<User> userMono = request.bodyToMono(User.class);
return userMono.flatMap( u->{
u.setCreatedDate(LocalDateTime.now());
return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
.body(userDao.saveAll(userMono), User.class);});
}
public Mono<ServerResponse> updateUser(ServerRequest request) {
Mono<User> userMono = request.bodyToMono(User.class);
String id = request.pathVariable("id");
return userMono.flatMap(updateUser -> {
return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
.body(userDao.findById(id).flatMap(oldUser -> {
// 将新的用户数据覆盖旧的用户数据
BeanUtils.copyProperties(updateUser, oldUser);
oldUser.setId(id);
oldUser.setCreatedDate(LocalDateTime.now());
return userDao.save(oldUser);
}), User.class);
});
}
public Mono<ServerResponse> deleteUserById (ServerRequest request) {
String id = request.pathVariable("id");
return userDao.findById(id)
.flatMap(u -> userDao.delete(u)
.then(ServerResponse.ok().build()))
.switchIfEmpty(ServerResponse.notFound().build());
}
}
```
配置Routers路由信息
```
@Configuration
public class UserRouters {
@Bean
RouterFunction<ServerResponse> userRouter(UserHandle handle){
return RouterFunctions.nest(
//相当于UserController里的 @RequestMapping("/user")
RequestPredicates.path("/userhandle"),
//相当于UserController里的 @GetMapping("/")
RouterFunctions.route(RequestPredicates.GET("/")
,handle::GetAll)
//相当于UserController里的 @GetMapping("/stream/all")
.andRoute(RequestPredicates.GET("/stream/all")
,handle::StreamGetAll)
//相当于UserController里的 @PostMapping("/")
.andRoute(RequestPredicates.POST("/")
.and(RequestPredicates.accept(MediaType.APPLICATION_JSON)),handle::createUser)
//相当于UserController里的 @PutMapping("/{id}")
.andRoute(RequestPredicates.PUT("/{id}")
.and(RequestPredicates.accept(MediaType.APPLICATION_JSON)),handle::updateUser)
//相当于UserController里的 @DeleteMapping("/{id}")
.andRoute(RequestPredicates.DELETE("/{id}"),handle::deleteUserById)
);
}
}
```
## 获取源码公众号回复: WebFlux


SpringBoot2.0学习之WebFlux