背景

场景	要求	编程方法
电商和金融行业	数据一致性要求非常高	高并发的时候需要锁或者其它机制来保证一些重要数据的一致性；但是性能也下降的很快；
游戏，新闻，视频，广告	不需要很高的数据一致性	对并发数和响应速度要求比较高

这种场景下，出现了响应式编程。依赖的基础技术点如下：

技术点	说明
servlet3.1	支持响应式编程
java8	语法丰富支持响应式编程,非堵塞式编程
spring5	新一代的web框架webflux，依托于servlet3.1 和java8
srpingboot2.x	使用了spring5
Rxjava	一种流行的响应式编程框架
Reactor	spring5中响应式编程的默认实现方式

基本概念

响应式编程关键词：

数据流：流式处理

异步：异步处理

消息：基于消息名

Reactor模型

1. 客户端先向服务器端注册感兴趣的event,完成了事件订阅；

1. 客户端发生已经注册的事件，会触发服务器的响应，服务器存在一个selector线程，【轮询客户端发送过来的事件】但是并不实际处理事件，而是找到对应的Request Handler，启用另外一条线程运行处理。

1. 最终结果会转换成data stream,发送到客户端；

WebFlux

基于servlet3.1对非阻塞机制，和java8的函数式语法，webflux出现了。

响应式编程分为3层：

层	说明
router functions	路由分发层，reactor模式中的selector; 根据请求的事件，选择对应的方法去处理客户端发送过来的事件请求。
spring webflux	控制层，处理业务逻辑前进行的封装和控制数据流的返回格式
http/reactive streams	转换层：把结果转换为数据流的过程

容器要求：支持servlet3.1

java异步编程领域：Netty

开发方式

开发方式	说明
类springMVC模式	简单，跟普通的springMVC方式有很多共同点，容易被接受
函数功能性	使用的比较少，因为不太好理解，开发后端的技术如果出现两种并存，效率不容易提高

数据流的封装

数据流封装	说明
Flux	存放0-N个数据流序列，响应式框架会一个一个的发送到客户端
Mono	存放0-1个数据流序列，一次仅发送一个数据流序列到客户端

DispatcherHandler

跟springMVC对标

对比	说明
DispatcherServlet	springMVC核心控制器
DispatcherHandler	webFlux核心处理器

核心处理代码

public Mono
   
     handle(ServerWebExchange exchange) {        if (this.handlerMappings == null) {            return createNotFoundError();        }        return Flux.fromIterable(this.handlerMappings)                .concatMap(mapping -> mapping.getHandler(exchange))                .next()                .switchIfEmpty(createNotFoundError())                .flatMap(handler -> invokeHandler(exchange, handler))                .flatMap(result -> handleResult(exchange, result));    }

核心处理流程

步骤	说明
1 DispatherHandler	接受请求
2 找到对应的HandlerMapping	从控制器中得到，通过@RequestMapping得到 fromIterable(this.handlerMappings)
3 得到对应的HandlerAdapter	.concatMap(mapping -> mapping.getHandler(exchange))
4 处理完毕之后得到Result	.flatMap(handler -> invokeHandler(exchange, handler))
5 返回到DispatherHandler
6 转换为handleResult	转换为对应的数据流 .flatMap(result -> handleResult(exchange, result));

注意

spring webflux只支持spring data reactive的数据源；

而数据库的开发往往是堵塞的，所以，spring data reactive并不能对数据库的开发提供支持。

适用于 redis,mongodb等nosql数据库

WebFlux的服务端开发

启动器代码

package com.springbootpractice.demo.webflux;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})@EnableReactiveMongoRepositories(basePackages = "com.springbootpractice.demo.webflux.dao.repository")public class DemoWebfluxApplication {    public static void main(String[] args) {        SpringApplication.run(DemoWebfluxApplication.class, args);    }}

持久层代码

package com.springbootpractice.demo.webflux.dao.repository;import com.springbootpractice.demo.webflux.dao.entity.User;import org.springframework.data.mongodb.repository.ReactiveMongoRepository;import org.springframework.stereotype.Repository;import reactor.core.publisher.Flux;/** * 说明：TODO * @author carter * 创建时间： 2020年01月15日 6:18 下午 **/@Repositorypublic interface UserRepository extends ReactiveMongoRepository
   
     {   Flux
    
      findByUserNameLikeAndNoteLike(String userName,String note);}

控制器代码

package com.springbootpractice.demo.webflux.controller;import com.springbootpractice.demo.webflux.core.UserValidator;import com.springbootpractice.demo.webflux.dao.entity.User;import com.springbootpractice.demo.webflux.service.UserService;import org.springframework.validation.DataBinder;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.InitBinder;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RestController;import reactor.core.publisher.Flux;import reactor.core.publisher.Mono;import javax.validation.Valid;/** * 说明：TODO * @author carter * 创建时间： 2020年01月15日 6:43 下午 **/@RestControllerpublic class UserController {    private final UserService userService;    public UserController(UserService userService) {        this.userService = userService;    }    @GetMapping(path = "/user/{id}")    public Mono
   
     getUser(@PathVariable("id") Long id){        return userService.getUserById(id);    }    @GetMapping(path = "/user/{userName}/{note}")    public Flux
    
      getUser(@PathVariable("userName") String userName, @PathVariable("note") String note){        return userService.findByUserNameAndNote(userName,note);    }    @GetMapping(path = "/user/insert/{user}")    public Mono
     
       insertUser(@Valid  @PathVariable("user")  User user){        return userService.insertUser(user);    }//    @InitBinder//    public void initBinder(DataBinder binder){//        binder.setValidator(new UserValidator());//    }}

WebFlux的核心功能

类比springMVC，提供了 WebFluxConfigurer进行配置，根据需要实现对应的方法；

转换器

对标 springMVC，也需要实现Converter接口：

代码

package com.springbootpractice.demo.webflux.core;import com.springbootpractice.demo.webflux.dao.entity.User;import com.springbootpractice.demo.webflux.dao.entity.enums.SexEnum;import org.springframework.core.convert.converter.Converter;import org.springframework.util.Assert;import java.util.Objects;/** * 说明：TODO * @author carter * 创建时间： 2020年01月16日 9:28 上午 **/public class String2UserConverter implements Converter
   
     {    @Override    public User convert(String s) {        final String[] split = Objects.requireNonNull(s,"转换为User的string不能为空").split("-");        Assert.isTrue(split.length==4,"转换为User的string必须含有4个字段");        return User.builder()                .id(Long.parseLong(split[0]))                .userName(split[1])                .note(split[2])                .sex(SexEnum.getSexEnum(Integer.parseInt(split[3])))                .build();    }}

校验器

对标springMVC的校验器，实现Validator接口；

代码

package com.springbootpractice.demo.webflux.core;import com.springbootpractice.demo.webflux.dao.entity.User;import org.apache.logging.log4j.util.Strings;import org.springframework.validation.Errors;import org.springframework.validation.Validator;/** * 说明：TODO * @author carter * 创建时间： 2020年01月16日 9:48 上午 **/public class UserValidator implements Validator {    @Override    public boolean supports(Class
    clazz) {        return clazz.equals(User.class);    }    @Override    public void validate(Object target, Errors errors) {        User user = (User) target;        if (Strings.isBlank(user.getUserName())){            errors.rejectValue("userName","","用户名不能为空");        }    }}

局部校验器

对标springMVC，可以在控制器中增加，里面配置好本控制器的校验器

代码

@InitBinder    public void initBinder(DataBinder binder){        binder.setValidator(new UserValidator());    }

设置静态资源

一些文件，图片，配置内容的配置，可以在WebConfigurer中进行配置；

代码

package com.springbootpractice.demo.webflux.core;import org.springframework.context.annotation.Configuration;import org.springframework.format.FormatterRegistry;import org.springframework.http.CacheControl;import org.springframework.validation.Validator;import org.springframework.web.reactive.config.ResourceHandlerRegistry;import org.springframework.web.reactive.config.WebFluxConfigurer;import java.util.concurrent.TimeUnit;/** * 说明：TODO * @author carter * 创建时间： 2020年01月16日 9:33 上午 **/@Configurationpublic class WebFluxConfig implements WebFluxConfigurer {    @Override    public void addFormatters(FormatterRegistry registry) {        registry.addConverter(new String2UserConverter());    }    @Override    public Validator getValidator() {        return new UserValidator();    }    @Override    public void addResourceHandlers(ResourceHandlerRegistry registry) {        registry.addResourceHandler("/static/**")                .addResourceLocations("classpath:/static/")                .setCacheControl(CacheControl.maxAge(1, TimeUnit.HOURS));    }}

访问静态资源例子：