Spring Webflux,spring-boot-starter-webflux

webflux介绍

Spring Boot 2.0

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

　　用弹簧靴做任何东西

　　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可以提高吞吐量和可伸缩性，但是接口的响应时间不会缩短，处理结果仍然要由工作线程处理后返回给请求。

　　00-1010适用于IO密集型、磁盘IO密集型、网络IO密集型等服务场景，如微服务网关。使用webflux技术可以显著提高网关对下游服务的吞吐量，spring cloud gateway使用了webflux技术。

　　00-1010要了解WebFlux，首先要了解什么是反应流。反应流是JVM中面向流的库的标准和规范；

　　它可以按顺序处理无限数量的元素，并在组件之间异步传输强制无阻塞背压Backpressure（背压）。

　　背压是一种常见的策略，它使发布者拥有无限的缓冲存储元素，并用于确保发布者在发布元素过快时不会抑制订阅者。

　　Reactive Streams（响应式流）

　　一般由以下组成：

　　通常包括以下内容：

　　发布者：发布者，向订阅者发布元素订阅者：订阅者，消费元素订阅：订阅，在发布者中，当创建订阅时，它将与订阅者共享处理器：处理器，发布者和订阅者之间的数据将被处理，这包括发布者和订阅者的社区publisher接口规范。

　　公共接口PublisherT { void subscribe(订户？超级T var 1)；}subscriber接口规范

　　公共接口subscriber { void on subscribe(Subscription var 1)；void on next(T var 1)；void on error(Throwable var 1)；void on complete()；}subscription接口规范

　　公共接口订阅{ void request(long var 1)；void cancel()；}processor接口规范

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

响应式编程

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

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

　　Mono：实现发布者，并返回 0 或 1 个元素Flux：实现发布者，并返回 N 个元素

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 生产的特性如下：

　　响应式 API编程模型适用性内嵌容器Starter 组件还有对日志、Web、消息、测试及扩展等支持。

　　响应式 API

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

spring webflux和spring mvc的异同点

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

　　MVC 能满足场景的，就不需要更改为 WebFlux。要注意容器的支持，可以看看下面内嵌容器的支持。微服务体系结构，WebFlux 和 MVC 可以混合使用。尤其开发 IO 密集型服务的时候，选择 WebFlux 去实现。spring mvc是一个命令式的编程方式采用同步阻塞方式，方便开发人员编写代码和调试；spring webflux调试会非常不方便JDBC连接池和JPA等技术还是阻塞模型，传统的关系型数据库如MySQL也不支持非阻塞的方式获取数据，目前只有非关系型数据库如Redis、Mongodb支持非阻塞方式获取数据编程模型

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

　　注解控制层。和 MVC 保持一致，WebFlux 也支持响应性 @RequestBody 注解。功能性端点。基于 lambda 轻量级编程模型，用来路由和处理请求的小工具。和上面最大的区别就是，这种模型，全程控制了请求 - 响应的生命流程内嵌容器

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

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

　　Netty优点

　　API使用简单、易上手功能强大、支持多种主流协议定制能力强、可扩展性高性能高、综合性能最优成熟稳定、久经考验社区活跃、学习资料多

Netty selector模型

Reactor指南

Reactor 框架是 Pivotal 公司（开发 Spring 等技术的公司）开发的实现了 Reactive Programming 思想，符合Reactive Streams 规范（Reactive Streams 是由 Netflix、TypeSafe、Pivotal 等公司发起的）的一项技术侧重于server端的响应式编程框架Reactor 框架主要有两个主要的模块：reactor-core 和 reactor-ipc。前者主要负责 Reactive Programming 相关的核心 API 的实现，后者负责高性能网络通信的实现，目前是基于 Netty 实现的。

Java原有的异步编程方式

Callback：异步方法采用一个callback作为参数，当结果出来后回调这个callback，例如swings的EventListenerFuture：异步方法返回一个Future<T>，此时结果并不是立刻可以拿到，需要处理结束之后才可以使用Future局限

　　多个Future组合不易调用Future#get时仍然会阻塞缺乏对多个值以及进一步的出错处理Reactor的Publisher

　　Mono 实现了 org.reactivestreams.Publisher 接口，代表0到1个元素的响应式序列。Flux 同样实现了 org.reactivestreams.Publisher 接口，代表0到N个元素的结果。Flux介绍

　　Flux<T>是一个标准Publisher<T>，表示0到N个发射项的异步序列，可选地以完成信号或错误终止。与Reactive Streams规范中一样，这三种类型的信号转换为对下游订阅者的onNext、onComplete或onError方法的调用。在这种大范围的可能信号中，Flux是通用的reactive 类型。注意，所有事件，甚至终止事件，都是可选的：没有onNext事件，但是onComplete事件表示一个空的有限序列，但是移除onComplete并且您有一个无限的空序列（除了关于取消的测试之外，没有特别有用）。同样，无限序列不一定是空的。例如，Flux.interval(Duration) 产生一个Flux<Long>，它是无限的，从时钟发出规则的数据。Mono介绍

　　Mono<T>是一个专门的Publisher<T>，它最多发出一个项，然后可选地以onComplete信号或onError信号结束。它只提供了可用于Flux的操作符的子集，并且一些操作符（特别是那些将Mono与另一个发布者组合的操作符）切换到Flux。例如，Mono#concatWith(Publisher)返回一个Flux ，而Mono#then(Mono)则返回另一个Mono。注意，Mono可以用于表示只有完成概念（类似于Runnable）的无值异步进程。若要创建一个，请使用Mono<Void>。publisher订阅

　　reactor实践

　　首先maven工厂引入pom

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

@RunWith(SpringRunner.class)@SpringBootTestpublic 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 - 元素映射为新元素

　　map操作可以将数据元素进行转换/映射，得到一个新元素。

　　flatMap - 元素映射为流

　　flatMap操作可以将每个数据元素转换/映射为一个流，然后将这些流合并为一个大的数据流。

　　filter - 过滤

　　filter操作可以对数据元素进行筛选。

　　zip - 一对一合并

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

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

　　用于编程方式自定义生成数据流的create和generate等及其变体方法；用于无副作用的peek场景的doOnNext、doOnError、doOncomplete、doOnSubscribe、doOnCancel等及其变体方法；用于数据流转换的when、and/or、merge、concat、collect、count、repeat等及其变体方法；用于过滤/拣选的take、first、last、sample、skip、limitRequest等及其变体方法；用于错误处理的timeout、onErrorReturn、onErrorResume、doFinally、retryWhen等及其变体方法；用于分批的window、buffer、group等及其变体方法；用于线程调度的publishOn和subscribeOn方法。使用这些操作符，你几乎可以搭建出能够进行任何业务需求的数据处理管道/流水线。

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

　　reactor和java8 stream区别

　　形似而神不似

　　reactor：push模式，服务端推送数据给客户端java8 stream：pull模式，客户端主动向服务端请求数据

Reactor线程模型

Reactor创建线程的方式

　　Schedulers.immediate()：当前线程Schedulers.single()：可重用的单线程，注意，这个方法对所有调用者都提供同一个线程来使用，直到该调度器被废弃。如果你想使用独占的线程，请使用Schedulers.newSingle()；Schedulers.elastic()：弹性线程池，它根据需要创建一个线程池，重用空闲线程。线程池如果空闲时间过长（默认为 60s）就会被废弃。对于 I/O 阻塞的场景比较适用。Schedulers.elastic()能够方便地给一个阻塞的任务分配它自己的线程，从而不会妨碍其他任务和资源；Schedulers.parallel()：固定大小线程池，所创建线程池的大小与CPU个数等同Schedulers.fromExecutorService(ExecutorService)：自定义线程池，基于自定义的ExecutorService创建 Scheduler（虽然不太建议，不过你也可以使用Executor来创建）线程模型

　　线程切换实践

@RunWith(SpringRunner.class)@SpringBootTestpublic 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); }}

线程切换总结

　　publishOn：它将上游信号传给下游，同时改变后续的操作符的执行所在线程，直到下一个publishOn出现在这个链上subscribeOn：作用于向上的订阅链，无论处于操作链的什么位置，它都会影响到源头的线程执行环境，但不会影响到后续的publishOn

webflux实践

兼容spring mvc的写法

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

spring webflux函数式写法

@Componentpublic 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); }}

配置路由

@Configurationpublic 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); }}

连接关系型数据库案例

@Componentpublic 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); }}

配置路由

@Configurationpublic 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属性

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

编写代码

@Document(collection = "user")@Datapublic class User { @Id private String id; private String name; private int age;} @Repositorypublic interface UserRepository extends ReactiveMongoRepository<User,String> {} @Componentpublic 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()); }}

编写路由

@Configurationpublic 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处理流程

　　具体步骤：

　　第一步：发起请求到前端控制器(DispatcherServlet)第二步：前端控制器请求HandlerMapping查找 Handler （可以根据xml配置、注解进行查找）匹配条件包括：请求路径、请求方法、header信息等第三步：处理器映射器HandlerMapping向前端控制器返回Handler，HandlerMapping会把请求映射为HandlerExecutionChain对象（包含一个Handler处理器（页面控制器）对象，多个HandlerInterceptor拦截器对象），通过这种策略模式，很容易添加新的映射策略HandlerInterceptor是请求路径上的拦截器，需要自己实现这个接口以拦截请求，做一些对handler的前置和后置处理工作。第四步：前端控制器调用处理器适配器去执行Handler第五步：处理器适配器HandlerAdapter将会根据适配的结果去执行Handler第六步：Handler执行完成给适配器返回ModelAndView第七步：处理器适配器向前端控制器返回ModelAndView （ModelAndView是springmvc框架的一个底层对象，包括 Model和view）第八步：前端控制器请求视图解析器去进行视图解析（根据逻辑视图名解析成真正的视图(jsp)），通过这种策略很容易更换其他视图技术，只需要更改视图解析器即可第九步：视图解析器向前端控制器返回View第十步：前端控制器进行视图渲染（视图渲染将模型数据(在ModelAndView对象中)填充到request域）第十一步：前端控制器向用户响应结果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); } } } ......}

webflux中引入了一个新的HandlerAdapter，即HandlerFunctionAdapterwebflux中引入了一个新的HandlerResultHandler，即ServerResponseResultHandlerServerResponseResultHandler实现了InitializingBean，因此在其实例化的时候，会调用afterPropertiesSet方法

　　流式处理请求handler()

@Overridepublic 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&      

	  




	  
	  
	  
	  
	  
        郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。