再看Spring之Spring5新特性——【Webflux】

1. Spring5新功能Spring5基于JDK8并且兼容JDK9。并且自带一个日志框架1.1 @Nullable注解可以添加在：属性，表示属性值可以为空；方法，返回值可以为空；参数，参数可以为空。1.2GenericApplicationContext函数式风格...

江海i

684人浏览 · 2022-05-31 11:23:27

江海i · 2022-05-31 11:23:27 发布

1. Spring5新功能

Spring5基于JDK8并且兼容JDK9。并且自带一个日志框架

1.1 `@Nullable`注解

可以添加在：

属性，表示属性值可以为空；
方法，返回值可以为空；
参数，参数可以为空。

1.2 `GenericApplicationContext`函数式风格

-注册对象

    public static void main(String[] args) {
        GenericApplicationContext context = new GenericApplicationContext();
        context.refresh();
        context.registerBean(User.class, User::new);
    }

获取对象

全路径类名：

        User user = (User) context.getBean("com.jm.spring.newInfo.User");
        System.out.println(user);

注册时指定对象名称：

       	GenericApplicationContext context = new GenericApplicationContext();
        context.refresh();
        context.registerBean("user1",User.class, User::new);
        User user1 = (User) context.getBean("user1");

1.3 Junit5 单元测试

// Spring测试类
@ExtendWith(SpringExtension.class)
// 指定配置文件
@ContextConfiguration("classpath:beans.xml")
public class MyTest2 {
    @Autowired
    private User user1;
    @Test
    public void Tg(){
        System.out.println(user1);
    }
 
}

复合注解：@SpringJunitConfig(localtions = "classpath:beans.xml")用来代替下方注解：

// Spring测试类
@ExtendWith(SpringExtension.class)
// 指定配置文件
@ContextConfiguration("classpath:beans.xml")

1.4 Webflux

1.4.1 基本概念

功能类似于SpringMVC，针对响应式编程的框架；相较于SpringMVC等传统MVC框架，WebFlux是一种异步非阻塞类型框架，基于Servlet3.1之后的框架，核心基于Reactor相关API实现。

异步非阻塞：
1. 异步：调用者发送请求之后无需等待回应就可以做出其它的操作；同步相反。
2. 阻塞：被调用者接收到请求，完成请求之后给出反馈，之后才做出其它操作；非阻塞：收到请求之后给出反馈，然后做其它操作。
特点：
1. 非阻塞式：在有限的资源下，提高系统的吞吐量和伸缩性，以Reactor为基础实现响应式编程；
2. 函数式编程：使用Java8函数式编程方式实现路由请求。
与SpringMVC的比较：
1. 都可以使用注解方式，运行在Tomcat等框架中；
2. SpringMVC命令式编程，Webflux异步响应式编程；

一般在网关中可以使用Webflux以提高系统的吞吐量。

1.4.2 响应式编程

是一种面向数据流和变化传播的编程范式。可以在语言中很方便地表达静态或者动态的数据流，相关的计算模型会自动将变化的值通过数据流进行传播。

简单的理解就是，如果通过A和B计算得出C的值，当A或者B修改时C的值也会随之变化。

Java8中通过Observer和Observable两个类实现观察者模式来达到响应式的目的

public class ObserverDemo extends Observable {
    public static void main(String[] args) {
        ObserverDemo observer = new ObserverDemo();

        // 添加观察者
        observer.addObserver((o, arg) -> {
            System.out.println("发生变化");
        });
        observer.addObserver((o, arg) -> {
            System.out.println("收到被观察者通知，准备改变。");
        });

        // 模拟发生数据变化
        observer.setChanged();

        // 发起通知
        observer.notifyObservers();
    }
}

1.4.3 Reactor

Reactor是满足Reacitve响应式编程规范的框架。

依赖：

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>3.1.5.RELEASE</version>
        </dependency>

Publisher接口： 提供Flux和Mono两个实现类。这两个实现类都是数据流的发布者，可以发出三种信号：
1. 元素值；
2. 错误信号；
3. 完成信号。
其中完成信号和错误信号都代表终止信号，不能共存，终止信号用于告诉订阅者数据流结束，但错误信号可以在终止数据流的同时返回错误信息个订阅者。

Flux类 ：实现发布者，返回 N个元素；

        List<Integer> list = new LinkedList<>();
        for (int i = 0; i < 5; i++) {
            list.add(i);
        }
        // 发送数据流 并且开启订阅
        Flux.fromIterable(list).subscribe(System.out::println);
        Flux.just(1,1,2,3,4).subscribe(System.out::println);

Mono类：实现发布者，返回 0 或 1 个元素。
```
Mono.just(1).subscribe(System.out::println);
```
信号特点：
1. 错误信号和完成信号不可以共存；
2. 没有发送任何元素，直接发生错误或完成信号，表示空数据流；
3. 没有发送错误信号或完成信号，表示无限数据流。
操作符：对数量流进行一道道操作，称为操作符号。
1. Map：将元素映射为新的元素，如下图中，对流中的元素进行操作并变成新的元素。
2. flatMap：将元素映射为流：将流中的元素经过某些操作之后，转为单独的流，并且把这些流合并到一个流中，返回这个统一的流。

1.4.4 netty

Webflux的默认容器是netty，而netty是高性能NIO，异步非阻塞框架。

NIO：每个操作都使用一个通道channel，把通道在选择器selector中进行注册，selector关注通道中的操作，实现多路复用方式。也就是异步非阻塞。

1.4.5 webflux执行过程

SpringWebflux的核心控制器是DispatcherHandler，实现接口WebHander。

WebHander接口

DispatcherHandler的方法实现

    /**
     * @param exchange http请求信息
     */
    public Mono<Void> handle(ServerWebExchange exchange) {
        // mapping空判断
        if (this.handlerMappings == null) {
            return createNotFoundError();
        }
        return Flux.fromIterable(this.handlerMappings)
                // 根据请求获取 mapping
                .concatMap(mapping -> mapping.getHandler(exchange))
                .next()
                .switchIfEmpty(createNotFoundError())
                // 执行业务具体方法
                .flatMap(handler -> invokeHandler(exchange, handler))
                // 返回处理结果
                .flatMap(result -> handleResult(exchange, result));
    }

DispatcherHandler中其它对象：
1. HanderMapping：请求查询到处理的方法；
2. HandeAdapter：处理请求；
3. HanderResultHander：响应处理结果。

1.4.6 webflux常见API

RouterFunction：路由实现：将请求转发给对应的handler。
HandlerFunction：处理请求，并且相应具体处理方法。

1.4.7 注解编程模型

新建SpringBoot项目，并修改maven配置文件如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.jm.webflux</groupId>
    <artifactId>web-flux</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>web-flux</name>
    <description>web-flux</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

模拟实体类：

/**
 * @Description
 * @date 2022/5/31 10:05
 */
public class User {

    private String name;

    private String gender;

    private Integer age;

    public User() {
    }

    public User(String name, String gender, Integer age) {
        this.name = name;
        this.gender = gender;
        this.age = age;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getGender() {
        return gender;
    }

    public void setGender(String gender) {
        this.gender = gender;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }
}

模拟逻辑层：

/**
 * @Description
 * @date 2022/5/31 10:06
 */
public interface UserService {

    /**
     * 根据id查询用户
     * @param id
     * @return
     */
    Mono<User> getUserById(Integer id);

    /**
     * 查询全部用户
     * @return
     */
    Flux<User> getAllUser();

    /**
     * 添加用户
     * @param userMono
     * @return
     */
    Mono<Void> saveUserInfo(Mono<User> userMono);

}

/**
 * @Description
 * @date 2022/5/31 10:09
 */
@Repository
public class UserServiceImpl implements UserService {

    // 模拟存储数据
    private final List<User> userList = new LinkedList<>();

    public UserServiceImpl(){
        userList.add(new User("zjm","男",22));
        userList.add(new User("lhy","女",21));
        userList.add(new User("my","1",18));
    }


    @Override
    public Mono<User> getUserById(Integer id) {
        return Mono.just(userList.get(id));
    }

    @Override
    public Flux<User> getAllUser() {
        return Flux.fromIterable(userList);
    }

    @Override
    public Mono<Void> saveUserInfo(Mono<User> userMono) {
        return userMono.doOnNext(userList::add).thenEmpty(Mono.empty());
    }
}

controller

/**
 * @Description
 * @date 2022/5/31 10:15
 */
@RestController
public class UserController {

    @Autowired
    private UserService userService;

    @GetMapping("/user/{id}")
    public Mono<User> getUserId(@PathVariable Integer id){
        return userService.getUserById(id);
    }

    @GetMapping("/user")
    public Flux<User> getAllUser(){
        return userService.getAllUser();
    }

    @PostMapping("/user/save")
    public Mono<Void> getUserId(@RequestBody User user){
        Mono<User> userMono = Mono.just(user);
        return userService.saveUserInfo(userMono);
    }


}

1.4.8 函数编程模型

同样是基于上方的项目实现，更改controller即可。

通过适配器将路由绑定到合适的handler
在这里插入图片描述

handler

/**
 * @Description
 * @date 2022/5/31 10:31
 */
public class UserHandler {

    // 获取逻辑层对象
    private final UserService userService;
    public UserHandler(UserService userService) {
        this.userService = userService;
    }

    /**
     * 根据id查询用户
     * @param request
     * @return
     */
    public Mono<ServerResponse> getUserId(ServerRequest request){
        // 获取查询出来的值
        Mono<User> userMono = this.userService.getUserById(Integer.valueOf(request.pathVariable("id")));

        // 空处理
        Mono<ServerResponse> notFound  = ServerResponse.notFound().build();

        // 使用 flatMap 操作符 返回
        return userMono.flatMap(user ->
                ServerResponse
                        .ok()
                        .contentType(MediaType.APPLICATION_JSON)
                        .body(fromObject(user)))
                .switchIfEmpty(notFound);
    }

    /**
     * 查询所有用户
     * @return
     */
    public Mono<ServerResponse> getAllUser(ServerRequest request){
        // 获取查询出来的值
        Flux<User> users = this.userService.getAllUser();
        // 空处理
        Mono<ServerResponse> notFound  = ServerResponse.notFound().build();

        // 使用 flatMap 操作符 返回
        return ServerResponse
                .ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(users,User.class);
    }

    /**
     * 添加用户
     * @param request
     * @return
     */
    public Mono<ServerResponse> saveUser(ServerRequest request){
        Mono<User> userMono = request.bodyToMono(User.class);
        return ServerResponse
                .ok()
                .build(this.userService.saveUserInfo(userMono));
    }
}

Server

/**
 * @Description 初始化服务器
 * @date 2022/5/31 10:52
 */
public class Server {

    public static void main(String[] args) {
        Server server = new Server();
        server.createReactorServer();
        System.out.println("enter to exit");
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建路由
     * @return
     */
    public RouterFunction<ServerResponse> routingFunction(){

        // 创建handler对象
        UserService userService = new UserServiceImpl();
        UserHandler handler = new UserHandler(userService);

        // 配置路径绑定方法
        return RouterFunctions
                .route(GET("/user/{id}").and(accept(MediaType.APPLICATION_JSON)), handler::getUserId)
                .andRoute(GET("/user").and(accept(MediaType.APPLICATION_JSON)), handler::getAllUser);
    }


    /**
     * 启动服务器 和 使用 adapter 绑定 路由 和 handler
     */
    public void createReactorServer(){
        // 适配 router 和 handler
        RouterFunction<ServerResponse> route = routingFunction();
        HttpHandler httpHandler = toHttpHandler(route);
        ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);

        // 初始化服务器
        HttpServer httpServer = HttpServer.create();
        httpServer.handle(adapter).bindNow();
    }

}

启动之后会有一个netty的端口，根据路由使用浏览器访问，或者将端口赋值到下方client中访问。
在这里插入图片描述

使用webClient调用

/**
 * @Description
 * @date 2022/5/31 11:16
 */
public class Client {

    public static void main(String[] args) {
        WebClient webClient = WebClient.create("http://127.0.0.1:53717");

        User user = webClient.get().
                uri("/user/{id}", 1)
                .accept(MediaType.APPLICATION_JSON)
                .retrieve().bodyToMono(User.class).block();
        System.out.println(user);

        Flux<User> userFlux = webClient.get().uri("/user")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve().bodyToFlux(User.class);

        userFlux.map(User::getName)
                .buffer().doOnNext(System.out::println).blockFirst();
    }

}