SpringBoot 2.x （14）：WebFlux响应式编程

Posted 2021-11-29 xuyiqing

响应式编程生活案例：

传统形式：

一群人去餐厅吃饭，顾客1找服务员点餐，服务员把订单交给后台厨师，然后服务员等待，

当后台厨师做好饭，交给服务员，经过服务员再交给顾客1，依此类推，该服务员再招待顾客2。

服务员可以理解为服务器，服务器越多，可处理的顾客请求越多

响应式编程：

服务员记住到顾客1的要求，交给后台厨师，再记住顾客2的要求，交给后台厨师，依此类推

当厨师做好顾客1的饭，告知服务员，然后服务员把饭送到顾客1；

当厨师做好顾客2的饭，告知服务员，然后服务员把饭送到顾客2，依此类推

 

一系列的事件称为流，异步非阻塞，观察者的设计模式

 

代码案例：

传统：

int b=2;
int c=3
int a=b+c //a被赋值后，b和c的改变不会影响a
b=5;

 

响应式编程：

int b=2;
int c=3
int a=b+c 
b=5;//此时a变化为8，a会根据b、c的变化而变化

 

SpringBoot2.x的响应式编程基于Spring5；

而Spring5的响应式编程又基于Reactor和Netty、Spring WebFlux替代Spring MVC

 

响应式编程最大的核心是非阻塞，即后台的每一步每一段都要做到非阻塞

比如使用mysql作为数据库，由于MySQL不提供响应式编程，所以会阻塞

因此响应式编程不应采用MySQL，应该使用非阻塞的NoSQL

 

Spring WebFlux有两种风格：基于功能和基于注解的。基于注解非常接近Spring MVC模型，如以下示例所示：

            @RestController 
            @RequestMapping（“/ users”）
             public  class MyRestController {

                @GetMapping（“/ {user}”）
                 public Mono <User> getUser（ @PathVariable Long user）{
                     // ...
                }

                @GetMapping（“/ {user} / customers”）
                 public Flux <Customer> getUserCustomers（ @PathVariable Long user）{
                     // ...
                }

                @DeleteMapping（“/ {user}”）
                 public Mono <User> deleteUser（ @PathVariable Long user）{
                     // ...
                }

            }

第二种： 路由配置与请求的实际处理分开

            @Configuration
             public  class RoutingConfiguration {

                @Bean
                 public RouterFunction <ServerResponse> monoRouterFunction（UserHandler userHandler）{
                     return route（GET（ “/ {user}”）.and（accept（APPLICATION_JSON）），userHandler :: getUser）
                            .andRoute（GET（“/ {user} / customers”）.and（accept（APPLICATION_JSON）），userHandler :: getUserCustomers）
                            .andRoute（DELETE（“/ {user}”）.and（accept（APPLICATION_JSON）），userHandler :: deleteUser）;
                }

            }

            @Component
            public class UserHandler {

                public Mono <ServerResponse> getUser（ServerRequest request）{
                     // ...
                }

                public Mono <ServerResponse> getUserCustomers（ServerRequest request）{
                     // ...
                }

                public Mono <ServerResponse> deleteUser（ServerRequest request）{
                     // ...
                }
            }

Spring WebFlux应用程序不严格依赖于Servlet API，因此它们不能作为war文件部署，也不能使用src/main/webapp目录

可以整合多个模板引擎，除了REST外，您还可以使用Spring WebFlux提供动态html内容

Spring WebFlux支持各种模板技术，包括Thymeleaf，FreeMarker

 

简单的实战：

依赖

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

自动生成的SpringBoot项目还会有一个test依赖，可选

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>

简单的Controller：

package org.dreamtech.webflux.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import reactor.core.publisher.Mono;

@RestController
public class TestController {
    @GetMapping("/test")
    public Mono<String> test() {
        return Mono.just("hello world");
    }
}

访问http://localhost:8080/test，显示hello world说明成功

 

这里使用到了Mono，后边还会用到Flux，他们的实现很复杂，但可以简单地理解：

User、List<User>

1）简单业务而言：和其他普通对象差别不大，复杂请求业务，就可以提升性能
2）通俗理解：
Mono 表示的是包含 0 或者 1 个元素的异步序列
mono->单一对象 User
例如从redis根据用户ID查到唯一的用户，然后进行返回Mono<User>

Flux 表示的是包含 0 到 N 个元素的异步序列
flux->数组列表对象 List<User>
例如从redis根据条件：性别为男性的用户进行查找，然后返回Flux<User>
3）Flux 和 Mono 之间可以进行转换

 

进一步的使用

对User实体类实现增删改查功能：

package org.dreamtech.webflux.domain;

public class User {
    private String id;
    private String name;

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

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

    public User(String id, String name) {
        super();
        this.id = id;
        this.name = name;
    }

}

Service：

package org.dreamtech.webflux.service;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.dreamtech.webflux.domain.User;
import org.springframework.stereotype.Service;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class UserService {
    // 使用Map模拟数据库
    private static final Map<String, User> dataMap = new HashMap<String, User>();
    static {
        dataMap.put("1", new User("1", "admin"));
        dataMap.put("2", new User("2", "John"));
        dataMap.put("3", new User("3", "Rose"));
        dataMap.put("4", new User("4", "James"));
        dataMap.put("5", new User("5", "Bryant"));
    }

    /**
     * 返回数据库的所有用户信息
     * 
     * @return
     */
    public Flux<User> list() {
        Collection<User> list = UserService.dataMap.values();
        return Flux.fromIterable(list);
    }

    /**
     * 根据用户ID返回用户信息
     * 
     * @param id 用户ID
     * @return
     */
    public Mono<User> getById(final String id) {
        return Mono.justOrEmpty(UserService.dataMap.get(id));
    }

    /**
     * 根据用户ID删除用户
     * 
     * @param id 用户ID
     * @return
     */
    public Mono<User> delete(final String id) {
        return Mono.justOrEmpty(UserService.dataMap.remove(id));
    }
}

Controller:

package org.dreamtech.webflux.controller;

import org.dreamtech.webflux.domain.User;
import org.dreamtech.webflux.service.UserService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
public class UserController {
    private final UserService userService;

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

    /**
     * 根据ID查找用户
     * 
     * @param id 用户ID
     * @return
     */
    @GetMapping("/find")
    public Mono<User> findById(final String id) {
        return userService.getById(id);
    }

    /**
     * 获得用户列表
     * 
     * @return
     */
    @GetMapping("/list")
    public Flux<User> list() {
        return userService.list();
    }

    /**
     * 根据ID删除用户
     * 
     * @param id 用户ID
     * @return
     */
    @GetMapping("/delete")
    public Mono<User> delete(final String id) {
        return userService.delete(id);
    }
}

访问定义的三个API，发现和SpringMVC基本没有区别

所以，对返回进行延迟处理：

    @GetMapping("/list")
    public Flux<User> list() {
        return userService.list().delayElements(Duration.ofSeconds(3));
    }

只是这些设置的话，等待3*list.size秒后全部返回，要突出流的特点，需要进行配置：

    @GetMapping(value = "/list", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<User> list() {
        return userService.list().delayElements(Duration.ofSeconds(3));
    }

这时候访问，可以发现每过3秒返回一个对象信息

 

使用WebClient客户端进行测试：

package org.dreamtech.webflux;

import org.junit.Test;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;

import reactor.core.publisher.Mono;

public class WebClientTest {
    @Test
    public void test() {
        Mono<String> bodyMono = WebClient.create().get().uri("http://localhost:8080/find?id=3")
                .accept(MediaType.APPLICATION_JSON).retrieve().bodyToMono(String.class);
        System.out.println(bodyMono.block());
    }
}