Spring5学习09:Webflux

川长思鸟来 2022-10-21 11:54 242阅读 0赞

## 1.SpringWebflux ##

（1）是 Spring5 添加新的模块，用于 web 开发的，功能和 SpringMVC 类似的，Webflux 使用当前一种比较流程响应式编程出现的框架。

![image.png][]

（2）使用传统 web 框架，比如 SpringMVC，这些基于 Servlet 容器，Webflux 是一种异步非阻塞的框架，异步非阻塞的框架在 Servlet3.1 以后才支持，核心是基于 Reactor 的相关 API 实现的。

(3)异步非阻塞

>  *  异步和同步针对调用者 异步和同步针对调用者，调用者发送请求，如果等着对方回应之后才去做其他事情就是同步，如果发送请求之后不等着对方回应就去做其他事情就是异步
>  *  阻塞和非阻塞针对被调用者 阻塞和非阻塞针对被调用者，被调用者受到请求之后，做完请求任务之后才给出反馈就是阻塞，受到请求之后马上给出反馈然后再去做事情就是非阻塞

（4）Webflux 特点：

>  *  非阻塞式：在有限资源下，提高系统吞吐量和伸缩性，以 Reactor 为基础实现响应式编程
>  *  函数式编程：Spring5 框架基于 java8，Webflux 使用 Java8 函数式编程方式实现路由请求

（5）比较 SpringMVC

![image.png][image.png 1]

>  *  两个框架都可以使用注解方式，都运行在 Tomet 等容器中
>  *  SpringMVC 采用命令式编程，Webflux 采用异步响应式编程

## 2.响应式编程（Java 实现） ##

### 1.什么是响应式编程 ###

> 响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。
> 
> 电子表格程序就是响应式编程的一个例子。单元格可以包含字面值或似"=B1+C1"的公式，而包含公式的单元格的值会依据其他单元格的值的变化而变化。

## 2.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(); //通知
        }
    }

## 3.响应式编程（Reactor 实现） ##

> 1.响应式编程操作中，Reactor 是满足 Reactive 规范框架
> 
> 2.Reactor 有两个核心类，**Mono** 和 和 **Flux** ，这两个类实现接口 Publisher，提供丰富操作 ，提供丰富操作。
> 
> Flux 对象实现发布者，返回 N 个元素；
> 
> Mono 实现发布者，返回 0 或者 1
> 
> 3.Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：
> 
> **元素值，错误信号，完成信号**
> 
> 错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者

![image.png][image.png 2]

4.代码

依赖

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

代码

public class TestReactor {
    
        public static void main(String[] args) {
            //just 方法直接声明
            Flux.just(1, 2, 3, 4, 5);
            Mono.just(1);
            
            //其它方法
            Integer[] array = {1,2,3,4};
            Flux.fromArray(array);//数组
            
            List<Integer> list = Arrays. asList (array);
            Flux.fromIterable(list);//集合
            
            Stream<Integer> stream = list.stream();
            Flux.fromStream(stream);//stream流
        }
    }

5.三种信号特点

>  *  错误信号和完成信号都是终止信号，不能共存的
>  *  如果没有发送任何元素值，而是直接发送错误或者完成信号，表示是空数据流
>  *  如果没有错误信号，没有完成信号，表示是无限数据流

6.调用 just 或者其他方法只是声明数据流，数据流并没有发出，只有进行订阅之后才会触发数据流，不订阅什么都不会发生的 subscribe

Flux.just(1, 2, 3, 4, 5).subscribe(System.out::println);
    Mono.just(1).subscribe(System.out::println);

7.操作符

对数据流进行一道道操作，成为操作符，比如工厂流水线

1.map 元素映射为新元素

![image.png][image.png 3]

2.flatMap 元素映射为流

把每个元素转换流，把转换之后多个流合并大的流

![image.png][image.png 4]

## 4.SpringWebflux 执行流程和核心 API ##

> SpringWebflux 基于 Reactor，默认使用容器是 Netty，Netty 是高性能的 NIO 框架，异步非阻塞的框架

### 1.Netty ###

BIO 阻塞

![image.png][image.png 5]

NIO 非阻塞

![image.png][image.png 6]

### 2.SpringWebflux 执行过程和 SpringMVC 相似的 ###

>  *  SpringWebflux 核心控制器 DispatchHandler，实现接口 WebHandler
>  *  接口 WebHandler 有一个方法

public interface WebHandler {
    
        /**
         * Handle the web server exchange.
         * @param exchange the current server exchange
         * @return {@code Mono<Void>} to indicate when request handling is complete
         */
        Mono<Void> handle(ServerWebExchange exchange);
    
    }

@Override
        public Mono<Void> handle(ServerWebExchange exchange) {
            if (this.handlerMappings == null) {
                return createNotFoundError();
            }
            if (CorsUtils.isPreFlightRequest(exchange.getRequest())) {
                return handlePreFlight(exchange);
            }
            return Flux.fromIterable(this.handlerMappings)
                    .concatMap(mapping -> mapping.getHandler(exchange))//获取对应的mapping
                    .next()
                    .switchIfEmpty(createNotFoundError())
                    .flatMap(handler -> invokeHandler(exchange, handler))//具体的方法
                    .flatMap(result -> handleResult(exchange, result)); //处理结果
        }

旧版代码

![image.png][image.png 7]

### 3.SpringWebflux 里面 DispatcherHandler，负责请求的处理 ###

>  *  HandlerMapping：请求查询到处理的方法
>  *  HandlerAdapter：真正负责请求处理
>  *  HandlerResultHandler：响应结果处理

### 4.SpringWebflux 实现函数式编程 ###

两个接口:

>  *  RouterFunction（路由处理）
>  *  HandlerFunction（处理函数）

## 5.SpringWebflux （基于注解编程模型） ##

> SpringWebflux 实现方式有两种：注解编程模型和函数式编程模型
> 
> 使用注解编程模型方式，和之前 SpringMVC 使用相似的，只需要把相关依赖配置到项目中，
> 
> SpringBoot 自动配置相关运行容器，默认情况下使用 Netty 服务器

1.创建springboot工程,可以在创建时勾选对应的web,自动导入依赖,也可以手动

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

2.配置启动端口号

server.port=8081

3.创建包和相关类

//实体类
    @Data
    @AllArgsConstructor
    public class User {
        private String name;
        private String gender;
        private Integer age;
    
    }
    
    
    //service接口和实现类
    public interface UserService {
    
        // 根据 id 查询用户
        Mono<User> getUserById(int id);
    
        // 查询所有用户
        Flux<User> getAllUser();
    
        // 添加用户
        Mono<Void> saveUserInfo(Mono<User> user);
    }
    @Repository
    public class UserServiceImpl implements UserService {
    
        //创建map集合存储数据
        private final Map<Integer,User> users = new HashMap<>();
    
        public UserServiceImpl() {
            this.users.put(1,new User("张三","男",33));
            this.users.put(2,new User("赛貂蝉","女",28));
            this.users.put(3,new User("胜西施","女",18));
        }
    
        //根据id查询
        @Override
        public Mono<User> getUserById(int id) {
            return Mono.justOrEmpty(this.users.get(id));
        }
    
        //查询多个用户
        @Override
        public Flux<User> getAllUser() {
            return Flux.fromIterable(this.users.values());
        }
    
        //添加用户
        @Override
        public Mono<Void> saveUserInfo(Mono<User> userMono) {
            return userMono.doOnNext(person -> {
                //向map集合里面放值
                int id = users.size()+1;
                users.put(id,person);
            }).thenEmpty(Mono.empty());
        }
    }
    
    @RestController
    public class UserController {
    
        //注入service
        @Autowired
        private UserService userService;
    
        //id查询
        @GetMapping("/user/{id}")
        public Mono<User> geetUserId(@PathVariable int id) {
            return userService.getUserById(id);
        }
    
        //查询所有
        @GetMapping("/user")
        public Flux<User> getUsers() {
            return userService.getAllUser();
        }
    
        //添加
        @PostMapping("/saveuser")
        public Mono<Void> saveUser(@RequestBody User user) {
            Mono<User> userMono = Mono.just(user);
            return userService.saveUserInfo(userMono);
        }
    }

说明

*  SpringMVC 方式实现，同步阻塞的方式，基于 SpringMVC+Servlet+Tomcat
 *  SpringWebflux 方式实现，异步非阻塞 方式，基于 SpringWebflux+Reactor+Netty

## 6.SpringWebflux （基于函数式编程模型） ##

> 1.在使用函数式编程模型操作时候，需要自己初始化服务器
> 
> 2.基于函数式编程模型时候，有两个核心接口：RouterFunction（实现路由功能，请求转发给对应的 handler）和 HandlerFunction（处理请求生成响应的函数）。核心任务定义两个函数式接口的实现并且启动需要的服务器。
> 
> 3.SpringWebflux 请 求 和 响 应 不 再 是 ServletRequest 和 ServletResponse ， 而 是
> 
> ServerRequest 和 ServerResponse

1.把注解编程模型工程复制一份 ，保留 entity 和 service 内容

2.创建 Handler（具体实现方法）

public class UserHandler {
    
        private final UserService userService;
        public UserHandler(UserService userService) {
            this.userService = userService;
        }
     //根据id查询
        public Mono<ServerResponse> getUserById(ServerRequest request) {
            //获取id值
           int userId = Integer.valueOf(request.pathVariable("id"));
           //空值处理
            Mono<ServerResponse> notFound = ServerResponse.notFound().build();
    
           //调用service方法得到数据
            Mono<User> userMono = this.userService.getUserById(userId);
            //把userMono进行转换返回
            //使用Reactor操作符flatMap
            return userMono.flatMap(person->
                ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(fromObject(person))
            )
            .switchIfEmpty(notFound);
            
        }
    
        //查询所有
        public Mono<ServerResponse> getAllUsers(ServerRequest request) {
            //调用service得到结果
            Flux<User> users = this.userService.getAllUser();
            return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(users,User.class);
        }
    
        //添加
        public Mono<ServerResponse> saveUser(ServerRequest request) {
            //得到user对象
            Mono<User> userMono = request.bodyToMono(User.class);
            return ServerResponse.ok().build(this.userService.saveUserInfo(userMono));
        }
    
    }

1.  初始化服务器，编写 Router

public class Server {
         //3.最终调用
        public static void main(String[] args) throws Exception{
            Server server = new Server();
            server.createReactorServer();
            System.out.println("enter to exit");
            System.in.read();
        }
    
        //1 创建Router路由
        public RouterFunction<ServerResponse> routingFunction() {
            //创建hanler对象
            UserService userService = new UserServiceImpl();
            UserHandler handler = new UserHandler(userService);
            //设置路由
            return RouterFunctions.route(
                    GET("/users/{id}").and(accept(APPLICATION_JSON)),handler::getUserById)
                    .andRoute(GET("/users").and(accept(APPLICATION_JSON)),handler::getAllUsers);
        }
    
        //2 创建服务器完成适配
        public void createReactorServer() {
            //路由和handler适配
            RouterFunction<ServerResponse> route = routingFunction();
            HttpHandler httpHandler = toHttpHandler(route);
            ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);
            //创建服务器
            HttpServer httpServer = HttpServer.create();
            httpServer.handle(adapter).bindNow();
        }
    }

4.使用 WebClient 调用

public class Client {
    
        public static void main(String[] args) {
            //调用服务器地址
            WebClient webClient = WebClient.create("http://127.0.0.1:56849");
    
            //根据id查询
            String id = "1";
            User userresult = webClient.get().uri("/users/{id}", id)
                    .accept(MediaType.APPLICATION_JSON).retrieve().bodyToMono(User.class)
                    .block();
            System.out.println(userresult.getName());
    
            //查询所有
            Flux<User> results = webClient.get().uri("/users")
                    .accept(MediaType.APPLICATION_JSON).retrieve().bodyToFlux(User.class);
    
            results.map(stu -> stu.getName())
                        .buffer().doOnNext(System.out::println).blockFirst();
        }
    }

[image.png]: /images/20221021/db7af9efc8324976a166b8c60542e798.png
[image.png 1]: /images/20221021/522f1c5aadc0471387ab5288f1f24be9.png
[image.png 2]: /images/20221021/684cbe93d78441a088117c07e43bc774.png
[image.png 3]: /images/20221021/b2af718d084b42d6a41e3680c1612cc4.png
[image.png 4]: /images/20221021/ba6bcf18a6a94f7099b41b25457c5cf9.png
[image.png 5]: /images/20221021/1afdfafa44254edc9259f73db232c7b1.png
[image.png 6]: /images/20221021/63a980f9dea7434687619cd653169bbe.png
[image.png 7]: /images/20221021/d57dd9362f7f431fbe7fcd0af2edac57.png