转载

0116 spring的webFlux

背景

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

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

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

基本概念

响应式编程关键词：

数据流：流式处理

异步：异步处理

消息：基于消息名

Reactor模型

客户端先向服务器端注册感兴趣的event,完成了事件订阅；
客户端发生已经注册的事件，会触发服务器的响应，服务器存在一个selector线程，【轮询客户端发送过来的事件】但是并不实际处理事件，而是找到对应的Request Handler，启用另外一条线程运行处理。
最终结果会转换成data stream,发送到客户端；

WebFlux

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

层	说明
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<Void> 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 下午
 **/
@Repository
public interface UserRepository extends ReactiveMongoRepository<User,Long> {

   Flux<User> 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 下午
 **/
@RestController
public class UserController {

    private final UserService userService;

    public UserController(UserService userService) {
        this.userService = userService;
    }

    @GetMapping(path = "/user/{id}")
    public Mono<User> getUser(@PathVariable("id") Long id){
        return userService.getUserById(id);
    }

    @GetMapping(path = "/user/{userName}/{note}")
    public Flux<User> getUser(@PathVariable("userName") String userName, @PathVariable("note") String note){
        return userService.findByUserNameAndNote(userName,note);
    }

    @GetMapping(path = "/user/insert/{user}")
    public Mono<User> 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<String, User> {
    @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 上午
 **/
@Configuration
public 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));

    }
}

访问静态资源例子：

0116 spring的webFlux