zhuxianfei.com

﹤首页 > 编程技术 > java

SpringBoot之webflux全面解析

发布时间：2022-2-24 14:21 作者：闻道☞

webflux介绍

Spring Boot 2.0

spring.io 官网有句醒目的话是：

BUILD ANYTHING WITH SPRING BOOT

Spring Boot （Boot 顾名思义，是引导的意思）框架是用于简化 Spring 应用从搭建到开发的过程。

应用开箱即用，只要通过一个指令，包括命令行 java -jar 、SpringApplication 应用启动类 、 Spring Boot Maven 插件等，就可以启动应用了。

另外，Spring Boot 强调只需要很少的配置文件，所以在开发生产级 Spring 应用中，让开发变得更加高效和简易。

目前，Spring Boot 版本是 2.x 版本。Spring Boot 包括 WebFlux。

传统的以SpringMVC为代表的webmvc技术使用的是同步阻塞式IO模型

而Spring WebFlux是一个异步非阻塞式IO模型，可以用少量的容器线程支撑大量的并发访问，所以Spring WebFlux可以提升吞吐量和伸缩性，但是接口的响应时间并不会缩短，其处理结果还是得由worker线程处理完成之后在返回给请求

webflux应用场景

适合IO密集型、磁盘IO密集、网络IO密集等服务场景，比如微服务网关，就可以使用webflux技术来显著的提升网关对下游服务的吞吐量，spring cloud gateway就使用了webflux这门技术

Spring Boot 2.0 WebFlux

了解 WebFlux，首先了解下什么是 Reactive Streams。Reactive Streams 是 JVM 中面向流的库标准和规范：

Backpressure（背压）

背压是一种常用策略，使得发布者拥有无限制的缓冲区存储元素，用于确保发布者发布元素太快时，不会去压制订阅者。

Reactive Streams（响应式流）

一般由以下组成：

一般由以下组成：

publisher接口规范

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> var1);
}

subscriber接口规范

public interface Subscriber<T> {
    void onSubscribe(Subscription var1); 
    void onNext(T var1); 
    void onError(Throwable var1); 
    void onComplete();
}

subscription接口规范

public interface Subscription {
    void request(long var1); 
    void cancel();
}

processor接口规范

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

响应式编程

有了 Reactive Streams 这种标准和规范，利用规范可以进行响应式编程。那再了解下什么是 Reactive programming 响应式编程。响应式编程是基于异步和事件驱动的非阻塞程序，只是垂直通过在 JVM 内启动少量线程扩展，而不是水平通过集群扩展。这就是一个编程范例，具体项目中如何体现呢？

响应式项目编程实战中，通过基于 Reactive Streams 规范实现的框架 Reactor 去实战。Reactor 一般提供两种响应式 API ：

Spring Webflux

Spring Boot Webflux 就是基于 Reactor 实现的。Spring Boot 2.0 包括一个新的 spring-webflux 模块。该模块包含对响应式 HTTP 和 WebSocket 客户端的支持，以及对 REST，HTML 和 WebSocket 交互等程序的支持。一般来说，Spring MVC 用于同步处理，Spring Webflux 用于异步处理。

Spring Boot Webflux 有两种编程模型实现，一种类似 Spring MVC 注解方式，另一种是使用其功能性端点方式。

Spring Boot 2.0 WebFlux 特性

常用的 Spring Boot 2.0 WebFlux 生产的特性如下：

还有对日志、Web、消息、测试及扩展等支持。

响应式 API

Reactor 框架是 Spring Boot Webflux 响应库依赖，通过 Reactive Streams 并与其他响应库交互。提供了 两种响应式 API：Mono 和 Flux。一般是将 Publisher 作为输入，在框架内部转换成 Reactor 类型并处理逻辑，然后返回 Flux 或 Mono 作为输出。

spring webflux和spring mvc的异同点

一图就很明确了，WebFlux 和 MVC 有交集，方便大家迁移。但是注意：

编程模型

Spring 5 web 模块包含了 Spring WebFlux 的 HTTP 抽象。类似 Servlet API , WebFlux 提供了 WebHandler API 去定义非阻塞 API 抽象接口。可以选择以下两种编程模型实现：

内嵌容器

跟 Spring Boot 大框架一样启动应用，但 WebFlux 默认是通过 Netty 启动，并且自动设置了默认端口为 8080。另外还提供了对 Jetty、Undertow 等容器的支持。开发者自行在添加对应的容器 Starter 组件依赖，即可配置并使用对应内嵌容器实例。

但是要注意，必须是 Servlet 3.1+ 容器，如 Tomcat、Jetty；或者非 Servlet 容器，如 Netty 和 Undertow。

Netty优点

Netty selector模型

Reactor指南

Java原有的异步编程方式

Future局限

Reactor的Publisher

Flux介绍

Mono介绍

publisher订阅

reactor实践

首先maven工厂引入pom

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

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor(){
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        Mono<Integer> mono = Mono.just(1);
 
        Integer[] arr = {1,2,3,4,5,6};
        Flux<Integer> flux1 = Flux.fromArray(arr);
 
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
        Flux<Integer> flux2 = Flux.fromIterable(list);
 
        Flux<Integer> flux3 = Flux.from(flux);
 
        Flux<Integer> flux4 = Flux.fromStream(Stream.of(1, 2, 3, 4, 5, 6));
 
        flux.subscribe();
 
        flux1.subscribe(System.out::println);
 
        flux2.subscribe(System.out::println,System.err::println);
 
        flux3.subscribe(System.out::println,System.err::println,() -> System.out.println("complete"));
 
        flux4.subscribe(System.out::println,System.err::println,
                () -> System.out.println("complete"),
                subscription -> subscription.request(3)); 
        flux4.subscribe(new DemoSubscriber());
    }
 
    class DemoSubscriber extends BaseSubscriber<Integer>{
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribe");
            subscription.request(1);
        }
 
        @Override
        protected void hookOnNext(Integer value) {
            if(value == 4){
                //背压，通知数据源，不要发送数据了
                cancel();
            }
            System.out.println(value);
            request(1);
        }
    }
}

Reactor操作符

map - 元素映射为新元素

flatMap - 元素映射为流

filter - 过滤

zip - 一对一合并

看到zip这个词可能会联想到拉链，它能够将多个流一对一的合并起来。zip有多个方法变体，我们介绍一个最常见的二合一的。

更多

Reactor中提供了非常丰富的操作符，除了以上几个常见的，还有：

使用这些操作符，你几乎可以搭建出能够进行任何业务需求的数据处理管道/流水线。

抱歉以上这些暂时不能一一介绍，更多详情请参考JavaDoc

reactor和java8 stream区别

形似而神不似

Reactor线程模型

Reactor创建线程的方式

线程模型

线程切换实践

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor() throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
 
        flux.map(i -> {
            System.out.println(Thread.currentThread().getName()+"-map1");
            return i * 3;
        }).publishOn(Schedulers.elastic()).map(
                i -> {
                    System.out.println(Thread.currentThread().getName()+"-map2");
                    return i / 3;
                }
        ).subscribeOn(Schedulers.parallel())
                .subscribe(i -> System.out.println(Thread.currentThread().getName()+"-" + i)); 
        Thread.sleep(10000);
    }
}

线程切换总结

webflux实践

兼容spring mvc的写法

@RestController
public class DemoController { 
    @GetMapping("/demo")
    public Mono<String> demo(){
        return Mono.just("demo");
    }
}

spring webflux函数式写法

@Component
public class DemoHandler { 
    public Mono<ServerResponse> hello(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("hello"),String.class);
    }
 
    public Mono<ServerResponse> world(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("world"),String.class);
    }
 
    public Mono<ServerResponse> times(ServerRequest request){
        //每隔一秒发送当前的时间
        return ok().contentType(MediaType.TEXT_EVENT_STREAM)
                .body(Flux.interval(Duration.ofSeconds(1))
                        .map(it -> new SimpleDateFormat("HH:mm:ss").format(new Date())),String.class);
    }
}

配置路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函数的编写
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times);
    }
}

连接关系型数据库案例

@Component
public class DemoHandler { 
    @Autowired
    private PersonService personService; 
    public Mono<ServerResponse> queryPerson(ServerRequest request){
        Integer id = Integer.valueOf(request.pathVariable("id"));
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(Mono.just(personService.getPersonById(id)), Person.class);
    }
}

配置路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函数的编写
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times)
                .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson);
    }
}

连接非关系型数据库案例

引入mongodb的maven

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

在application.properties中配置mongodb属性

#mongodb
spring.data.mongodb.uri=mongodb://root:yibo@localhost:27017
spring.data.mongodb.database=webflux

编写代码

@Document(collection = "user")
@Data
public class User { 
    @Id
    private String id; 
    private String name; 
    private int age;
}
 
@Repository
public interface UserRepository extends ReactiveMongoRepository<User,String> {
}
 
@Component
public class DemoHandler { 
    @Autowired
    private UserRepository userRepository; 
    public Mono<ServerResponse> listUser(ServerRequest request){
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(userRepository.findAll(), User.class);
    }
 
    public Mono<ServerResponse> saveUser(ServerRequest request){
        String name = request.pathVariable("name");
        Integer age = Integer.valueOf(request.pathVariable("age"));
        User user = new User();
        user.setName(name);
        user.setAge(age);
        Mono<User> mono = Mono.just(user);
        return ok().build(userRepository.insert(mono).then());
    }
}

编写路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函数的编写
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times)
                .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson)
                .andRoute(GET("/listUser"),demoHandler::listUser)
                .andRoute(GET("/saveUser/{name}/{age}"),demoHandler::saveUser);
    }
}

webflux解析

spring mvc处理流程

具体步骤：

spring webflux处理请求流程

核心控制器DispatcherHandler，等同于阻塞方式的DispatcherServlet

DispatcherHandler实现ApplicationContextAware，那么必然会调用setApplicationContext方法

public class DispatcherHandler implements WebHandler, ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        initStrategies(applicationContext);
    }
}

initStrategies初始化

获取HandlerMapping，HandlerAdapter，HandlerResultHandler的所有实例

protected void initStrategies(ApplicationContext context) {
    //获取HandlerMapping及其子类型的bean
    //HandlerMapping根据请求request获取handler执行链
    Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerMapping.class, true, false);
 
    ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(mappings);
    this.handlerMappings = Collections.unmodifiableList(mappings);
 
    //获取HandlerAdapter及其子类型的bean
    Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerAdapter.class, true, false);
 
    this.handlerAdapters = new ArrayList<>(adapterBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(this.handlerAdapters);
 
    //获取HandlerResultHandler及其子类型的bean
    Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerResultHandler.class, true, false);
 
    this.resultHandlers = new ArrayList<>(beans.values());
    AnnotationAwareOrderComparator.sort(this.resultHandlers);
}

webflux中引入了一个新的HandlerMapping，即RouterFunctionMapping

RouterFunctionMapping实现了InitializingBean，因此在其实例化的时候，会调用afterPropertiesSet方法

public class RouterFunctionMapping extends AbstractHandlerMapping implements InitializingBean {
 
    @Nullable
    private RouterFunction<?> routerFunction;
 
    //读取http传输数据，并解码成一个对象
    private List<HttpMessageReader<?>> messageReaders = Collections.emptyList();
 
    public RouterFunctionMapping(RouterFunction<?> routerFunction) {
        this.routerFunction = routerFunction;
    }
 
    @Nullable
    public RouterFunction<?> getRouterFunction() {
        return this.routerFunction;
    }
 
    public void setMessageReaders(List<HttpMessageReader<?>> messageReaders) {
        this.messageReaders = messageReaders;
    }
 
    @Override
    public void afterPropertiesSet() throws Exception {
        if (CollectionUtils.isEmpty(this.messageReaders)) {
            ServerCodecConfigurer codecConfigurer = ServerCodecConfigurer.create();
            this.messageReaders = codecConfigurer.getReaders();
        }
 
        if (this.routerFunction == null) {
            //afterPropertiesSet方法调用的时候，routerFunction为null
            initRouterFunctions();
        }
    }
 
    protected void initRouterFunctions() {
        //获取routerFunctions集合
        List<RouterFunction<?>> routerFunctions = routerFunctions();
         //将一个请求中含有多个路由请求RouterFunction合并成一个RouterFunction
        this.routerFunction = routerFunctions.stream().reduce(RouterFunction::andOther).orElse(null);
        logRouterFunctions(routerFunctions);
    }
 
    private List<RouterFunction<?>> routerFunctions() {
        //obtainApplicationContext()获取ApplicationContext对象
        List<RouterFunction<?>> functions = obtainApplicationContext()
                //获取指定bean的提供者，即上文配置的路由类
                .getBeanProvider(RouterFunction.class)
                //排序
                .orderedStream()
                //将流里面的都强转成RouterFunction对象
                .map(router -> (RouterFunction<?>)router)
                .collect(Collectors.toList());
        return (!CollectionUtils.isEmpty(functions) ? functions : Collections.emptyList());
    }
 
    private void logRouterFunctions(List<RouterFunction<?>> routerFunctions) {
        //判断当前的日志级别是否是Debug
        if (logger.isDebugEnabled()) {
            int total = routerFunctions.size();
            String message = total + " RouterFunction(s) in " + formatMappingName();
            if (logger.isTraceEnabled()) {
                if (total > 0) {
                    routerFunctions.forEach(routerFunction -> logger.trace("Mapped " + routerFunction));
                }
                else {
                    logger.trace(message);
                }
            }
            else if (total > 0) {
                logger.debug(message);
            }
        }
    }
    ......
}

ServerResponseResultHandler实现了InitializingBean，因此在其实例化的时候，会调用afterPropertiesSet方法

流式处理请求handler()

@Override
public Mono<Void> handle(ServerWebExchange exchange) {
    //handlerMappings在initStrategies()方法中已经构造好了
    if (this.handlerMappings == null) {
        return createNotFoundError();
    }
    //构造Flux，数据源为handlerMappings集合
    return Flux.fromIterable(this.handlerMappings)
            //获取Mono<Handler>对象，通过concatMap保证顺序和handlerMappings顺序一致
            //严格保证顺序是因为在一个系统中可能存在一个Url有多个能够处理的HandlerMapping的情况
            .concatMap(mapping -> mapping.getHandler(exchange))
            .next()
            //如果next()娶不到值则抛出错误
            .switchIfEmpty(createNotFoundError())
            //触发HandlerApter的handle方法
            .flatMap(handler -> invokeHandler(exchange, handler))
            //触发HandlerResultHandler 的handleResult方法
            .flatMap(result -> handleResult(exchange, result));
}

触发HandlerApter的handle方法

private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
    return getResultHandler(result).handleResult(exchange, result)
            .onErrorResume(ex -> result.applyExceptionHandler(ex).flatMap(exceptionResult ->
                    getResultHandler(exceptionResult).handleResult(exchange, exceptionResult)));
}
 
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
    if (this.resultHandlers != null) {
        for (HandlerResultHandler resultHandler : this.resultHandlers) {
            if (resultHandler.supports(handlerResult)) {
                return resultHandler;
            }
        }
    }
    throw new IllegalStateException("No HandlerResultHandler for " + handlerResult.getReturnValue());
}

总结

DispatcherHandler的流程是

1、通过 HandlerMapping（和DispathcherServlet中的HandlerMapping不同）获取到HandlerAdapter放到ServerWebExchange的属性中

2、获取到HandlerAdapter后触发handle方法，得到HandlerResult3、通过HandlerResult，触发handleResult，针对不同的返回类找到不同的HandlerResultHandler如视图渲染ViewResolutionResultHandler、ServerResponseResultHandler、ResponseBodyResultHandler、ResponseEntityResultHandler不同容器有不同的实现，如Reactor，Jetty，Tomcat等。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持猪先飞。

原文出处：https://blog.csdn.net/u014401141/article/details/107430161

标签：[!--infotagslink--]

您可能感兴趣的文章: