提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其他

当前栏目

Spring5---新特性(WebFlux)

编程
2023-03-14 22:35:16 时间

WebFlux

SpringWebflux介绍

Spring WebFlux是随Spring 5推出的响应式Web框架。

传统的web框架，比如springmvc，这些是基于servlet容器，webflux是一种异步非阻塞的框架，异步非阻塞的框架是在servlet 3.1 以后才支持的，核心是基于Reactor的相关API实现的

Webflux特点

非阻塞式:在有限资源下，提高系统吞吐量和伸缩性，以Reactor为基础实现响应式编程

函数式编程,spring5框架基于java8,Webflux使用java8函数式编程方式实现路由请求

SpringMvc和Webflux进行比较

两个框架都可以使用注解的方式，都运行在Tomcat等容器中

SpringMvc采用命令式编程，WebFlux采用异步响应式编程

响应式编程

响应式编程是一种面向数据流和变化传播的编程范式。对于响应式框架，是基于响应式宣言的理念所产生的编程方式。响应式宣言分为4大理念：

  • 灵敏的：可以快速响应的，只要有任何可能，系统都应该能够尽可能快地做出响应。
  • 可恢复的：系统在运行中可能出现问题，但是能够有很强大的容错机制和修复机制保持响应性。
  • 可伸缩的：在任何负载下，响应式编程都可以根据自身压力变化，请求少时，通过减少资源释放服务器压力，负载大时能够通过扩展算法和软硬件的方式扩展服务压力，以经济实惠的方式实现可伸缩性
  • 消息驱动的：响应式编程存在异步消息机制，事件之间的协作是通过消息进行连接的。

基于这些理念，响应式编程提出了各种模型来满足响应式编程的理念，其中著名的有Reactor和RxJava，Spring5就是基于它们构建WebFlux，而默认情况下它会使用Reactor。

JAVA代码演示

Java8及其之前版本，提供的观察者模式的两个类:Observer和Observerable

public class Observe extends Observable
{
    public static void main(String[] args) {
        Observe observer=new Observe();
        //添加观察者
        observer.addObserver((o,arg)->{
            System.out.println("发生变化");
        });
        observer.addObserver((o,arg)->{
            System.out.println("手动被观察者通知，准备改变");
        });
        observer.setChanged();//数据变化
        observer.notifyObservers();//通知
    }
}

响应式编程(Reactor实现)

1.响应式编程操作中，Reactor是满足Reactive规范框架

2.Reactor有两个核心类，Mono和Flux，这两个类实现接口Publisher，提供丰富操作，Flux对象实现发布者，返回N个元素; Mono实现发布者，返回0或者1个元素

3.Flux和Mono都是数据流的发布者，使用Flux和Mono都可以发出三种数据信号：元素值，错误信号，完成信号; 错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者

代码演示Flux和Mono

首先导入Reactor核心包的依赖:

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

编写代码:

        //just方法直接声明相关元素
     Flux.just(1,2,3,4);
     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);

三种信号特点:

错误信号和完成信号都是终止信号，两个不能共存

如果没有发送任何元素值，而是直接发送错误或者完成新型，表示是空数据流

如果没有错误信号，没有完成信号，表示是无限的数据流

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

        //just方法直接声明相关元素
        //subscribe(Consumer<? super T> consumer)
        //可以实现该函数式接口
     Flux.just(1,2,3,4).subscribe(System.out::println);//订阅
     Mono.just(1).subscribe(System.out::println);

操作符

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

map 元素映射为新元素

flatmap元素映射为流，每个元素转换为流，把转换之后的多个流合并为一个大流返回

SpringWebFlux执行流程和核心API

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

(1)Netty

BIO:阻塞方式

NIO：

SpringWebflux与SpringWebMvc是相似的

引入webflux对应的启动器

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

SpringWebflux核心控制器DispatchHandler,实现接口WebHandler

DispatchHandler负责请求的派发

handlerMapping:请求查询到处理的方法

HandlerAdapter:真正负责请求处理

handleResultHandler:响应结果处理

SpingWebflux实现函数式编程，两个接口： RouterFunction(路由的处理)和HandlerFunction（处理函数）

SpringWebflux(基于注解编程模型)

SpringWebflux实现方式有两种: 注解编程模型和函数式编程模型

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

步骤1 ：创建springboot项目，引入上面给出的webflux启动器

步骤2：配置启动端口号

server.port=8081

步骤3：创建实体类(Lombok插件)

@Data
@AllArgsConstructor
@NoArgsConstructor
public class User
{
    String name;
    Integer age;
}

步骤4：创建service接口

UserService：

//用户操作接口
public interface UserService
{
 //根据id查询用户
    //Mono返回0个或1个元素
    public Mono<User>  getUserById(Integer id);
    //查询所有用户
    public Flux<User> getAll();
    //添加用户
    public  Mono<Void> saveUserInfo( Mono<User> userMono);
}

步骤5： 创建service接口的实现类

@Repository
public class UserServiceImpi implements UserService
{
    //创建map集合存储数据---模拟数据库操作
    private  final Map<Integer,User> users=new HashMap<>();

    UserServiceImpi()
    {
     users.put(1,new User("大忽悠1", 18));
     users.put(2,new User("大忽悠2", 19));
     users.put(3,new User("大忽悠3", 20));
    }
    //根据id查询
    @Override
    public Mono<User> getUserById(Integer id) {
        return Mono.justOrEmpty(users.get(id));
    }

    @Override
    public Flux<User> getAll() {
        //返回一个集合
        return Flux.fromIterable(users.values());
    }

    //模拟保存用户到数据库
    @Override
    public Mono<Void> saveUserInfo(Mono<User> userMono) {
       //这里提供一个Consumer函数式接口，即提供给我们一个User类的对象，我们可以对该对象进行消费处理
        return userMono.doOnNext(u->{
          //往map集合里面放值
            int id=users.size()+1;
            users.put(id,u);
      }).thenEmpty(Mono.empty());//结束数据流(终止信号)
    }
}

步骤6：创建controller层

@RestController
public class UserController
{
    //注入service
    @Autowired
    private UserService userService;

    //id查询
    @GetMapping("/user/{id}")
    public Mono<User> getUserById(@PathVariable Integer id)
    {
        return userService.getUserById(id);
    }
    //查询所有
    @GetMapping("/user")
            public Flux<User> getUsers()
    {
        return userService.getAll();
    }
    //添加
    @PostMapping("/saveuser")
    //@RequestBody:将前端的json字符串封装为指定对象
    public Mono<Void> saveUser(@RequestBody User user)
    {
           Mono<User> userMono=Mono.just(user);
           return userService.saveUserInfo(userMono);
    }
}

步骤7：测试

说明:

SpringMVC方式实现，同步阻塞的方式，基于SpringMVC+Servlet+Tomcat

SrpingWebflux方式实现，异步非阻塞方式，基于SpringWebflux+Reactor+Netty

SpringWebflux(基于函数式编程模型)

1.在使用函数式编程模型操作的时候，需要自己初始化服务器

2.基于函数式编程模式的时候，有两个核心接口

RouterFunction(实现路由功能，请求转发给对应的handler)

HandlerFunction(处理请求生成响应的函数)

核心任务定义两个函数式接口的实现并且启动需要的服务器

3.SpringWebflux请求和响应不在是ServletRequest和ServletResponse

而是ServerRequest和ServerResponse

步骤1 : 将上面注解版本的项目复制一份，并删除注解版本里面的controller层

步骤2：controller层替换为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 = userService.getAll();
                                                                           //不同的写法
        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));
    }
}

步骤3：初始化服务器，编写Router

创建路由

public class Server
{
    //1.创建router路由
    public RouterFunction<ServerResponse> responseRouterFunction()
    {
        //创建handler对象
        UserService userService = new UserServiceImpi();
        UserHandler handler = new UserHandler(userService);
        //设置路由
        return RouterFunctions.
                route(GET("/user/{id}").
                        and(accept(APPLICATION_JSON)), handler::getUserById).
                andRoute(GET("/user").and(accept(APPLICATION_JSON)), handler::getAllUsers);
    }
}

创建服务器完成适配

    //创建服务器完成适配
    public void createReactorServer()
    {
        //路由和Hander适配
        RouterFunction<ServerResponse> route=responseRouterFunction();
        HttpHandler httpHandler=  toHttpHandler(route);
        ReactorHttpHandlerAdapter adapter=new ReactorHttpHandlerAdapter
                (httpHandler);

        //创建服务器
        HttpServer httpServer=HttpServer.create();
        httpServer.handle(adapter).bindNow();
    }

最终调用

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

完整的服务器类代码

public class Server
{
    public static void main(String[] args) throws IOException {
        Server server=new Server();
        server.createReactorServer();
        System.out.println("enter to exit");
        System.in.read();
    }
    //1.创建router路由
    public RouterFunction<ServerResponse> responseRouterFunction()
    {
        //创建handler对象
        UserService userService = new UserServiceImpi();
        UserHandler handler = new UserHandler(userService);
        //设置路由
        return RouterFunctions.
                route(GET("/user/{id}").
                        and(accept(APPLICATION_JSON)), handler::getUserById).
                andRoute(GET("/user").and(accept(APPLICATION_JSON)), handler::getAllUsers);
    }
    //创建服务器完成适配
    public void createReactorServer()
    {
        //路由和Hander适配
        RouterFunction<ServerResponse> route=responseRouterFunction();
        HttpHandler httpHandler=  toHttpHandler(route);
        ReactorHttpHandlerAdapter adapter=new ReactorHttpHandlerAdapter
                (httpHandler);

        //创建服务器
        HttpServer httpServer=HttpServer.create();
        httpServer.handle(adapter).bindNow();
    }
}

测试:

使用WebCilent进行调用

public class Cilent
{
    public static void main(String[] args) {
        //调用服务器之前，对应服务器要先启动起来，因为默认使用随机的端口号，因此我们每一次启动对应的服务器
        //下次再连接时，都需要对端口号进行修改
        //调用服务器的地址
        WebClient webClient = WebClient.create("http://localhost:61546");
        //根据id查询
        String id="1";
        User user = webClient.get().uri("/user/{id}",id).accept(MediaType.APPLICATION_JSON)
                .retrieve().bodyToMono(User.class).block();
        System.out.println(user.getName());
        //查询所有
        Flux<User> userFlux = webClient.get().uri("/user").accept(MediaType.APPLICATION_JSON).
                retrieve().bodyToFlux(User.class);
        userFlux.map(s->s.getName()).buffer().doOnNext(System.out::println).blockFirst();
    }
}

相关文章