每日一博 | 深入剖析 Spring WebFlux
一、WebFlux 简介
WebFlux 是 Spring Framework5.0 中引入的一种新的反应式Web框架。通过Reactor项目实现Reactive Streams规范,完全异步和非阻塞框架。本身不会加快程序执行速度,但在高并发情况下借助异步IO能够以少量而稳定的线程处理更高的吞吐,规避文件IO/网络IO阻塞带来的线程堆积。
1.1 WebFlux 的特性
WebFlux 具有以下特性:
-
异步非阻塞 - 可以举一个上传例子。相对于 Spring MVC 是同步阻塞IO模型,Spring WebFlux这样处理:线程发现文件数据没传输好,就先做其他事情,当文件准备好时通知线程来处理(这里就是输入非阻塞方式),当接收完并写入磁盘(该步骤也可以采用异步非阻塞方式)完毕后再通知线程来处理响应(这里就是输出非阻塞方式)。
-
响应式函数编程 - 相对于Java8 Stream 同步、阻塞的Pull模式,Spring Flux 采用Reactor Stream 异步、非阻塞Push模式。书写采用 Java lambda 方式,接近自然语言形式且容易理解。
-
不拘束于Servlet - 可以运行在传统的Servlet 容器(3.1+版本),还能运行在Netty、Undertow等NIO容器中。
1.2 WebFlux 的设计目标
-
适用高并发
-
高吞吐量
-
可伸缩性
二、Spring WebFlux 组件介绍
2.1 HTTPHandler
一个简单的处理请求和响应的抽象,用来适配不同HTTP服务容器的API。
2.2 WebHandler
一个用于处理业务请求抽象接口,定义了一系列处理行为。相关核心实现类如下;
2.3 DispatcherHandler
请求处理的总控制器,实际工作是由多个可配置的组件来处理。
WebFlux是兼容Spring MVC 基于@Controller,@RequestMapping等注解的编程开发方式的,可以做到平滑切换。
2.4 Functional Endpoints
这是一个轻量级函数编程模型。是基于@Controller,@RequestMapping等注解的编程模型的替代方案,提供一套函数式API 用于创建Router,Handler和Filter。调用处理组件如下:
简单的RouterFuntion 路由注册和业务处理过程:
@Bean public RouterFunction<ServerResponse> initRouterFunction() { return RouterFunctions.route() .GET("/hello/{name}", serverRequest -> { String name = serverRequest.pathVariable("name"); return ServerResponse.ok().bodyValue(name); }).build(); }
请求转发处理过程:
2.5 Reactive Stream
这是一个重要的组件,WebFlux 就是利用Reactor 来重写了传统Spring MVC 逻辑。其中Flux和Mono 是Reactor中两个关键概念。掌握了这两个概念才能理解WebFlux工作方式。
Flux和Mono 都实现了Reactor的Publisher接口,属于时间发布者,对消费者提供订阅接口,当有事件发生的时候,Flux或者Mono会通过回调消费者的相应的方法来通知消费者相应的事件。这就是所谓的响应式编程模型。
Mono工作流程图
只会在发送出单个结果后完成。
Flux工作流程图
发送出零个或者多个,可能无限个结果后才完成。
对于流式媒体类型:application/stream+json 或者 text/event-stream ,可以让调用端获得服务器滚动结果。 对于非流类型:application/json WebFlux 默认JSON编码器会将序列化的JSON 一次性刷新到网络,这并不意味着阻塞,因为结果Flux<?> 是以反应式方式写入网络的,没有任何障碍。
三、WebFlux 工作原理
3.1 组件装配过程
流程相关源码解析-WebFluxAutoConfiguration
@Configuration //条件装配 只有启动的类型是REACTIVE时加载 @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE) //只有存在 WebFluxConfigurer实例 时加载 @ConditionalOnClass(WebFluxConfigurer.class) //在不存在 WebFluxConfigurationSupport实例时 加载 @ConditionalOnMissingBean({ WebFluxConfigurationSupport.class }) //在之后装配 @AutoConfigureAfter({ ReactiveWebServerFactoryAutoConfiguration.class, CodecsAutoConfiguration.class, ValidationAutoConfiguration.class }) //自动装配顺序 @AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE + 10) public class WebFluxAutoConfiguration { @Configuration @EnableConfigurationProperties({ ResourceProperties.class, WebFluxProperties.class }) //接口编程 在装配WebFluxConfig 之前要先 装配EnableWebFluxConfiguration @Import({ EnableWebFluxConfiguration.class }) public static class WebFluxConfig implements WebFluxConfigurer { //隐藏部分源码 /** * Configuration equivalent to {@code @EnableWebFlux}. */ } @Configuration public static class EnableWebFluxConfiguration extends DelegatingWebFluxConfiguration { //隐藏部分代码 } @Configuration @ConditionalOnEnabledResourceChain static class ResourceChainCustomizerConfiguration { //隐藏部分代码 } private static class ResourceChainResourceHandlerRegistrationCustomizer implements ResourceHandlerRegistrationCustomizer { //隐藏部分代码 }
WebFluxAutoConfiguration 自动装配时先自动装配EnableWebFluxConfiguration 而EnableWebFluxConfiguration->DelegatingWebFluxConfiguration ->WebFluxConfigurationSupport。
最终WebFluxConfigurationSupport 不仅配置DispatcherHandler 还同时配置了其他很多WebFlux核心组件包括 异常处理器WebExceptionHandler,映射处理器处理器HandlerMapping,请求适配器HandlerAdapter,响应处理器HandlerResultHandler 等。
DispatcherHandler 创建初始化过程如下;
public class WebFluxConfigurationSupport implements ApplicationContextAware { //隐藏部分代码 @Nullable public final ApplicationContext getApplicationContext() { return this.applicationContext; } //隐藏部分代码 @Bean public DispatcherHandler webHandler() { return new DispatcherHandler(); }
public class DispatcherHandler implements WebHandler, ApplicationContextAware { @Nullable private List<HandlerMapping> handlerMappings; @Nullable private List<HandlerAdapter> handlerAdapters; @Nullable private List<HandlerResultHandler> resultHandlers; @Override public void setApplicationContext(ApplicationContext applicationContext) { initStrategies(applicationContext); } protected void initStrategies(ApplicationContext context) { //注入handlerMappings 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); //注入handlerAdapters Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors( context, HandlerAdapter.class, true, false); this.handlerAdapters = new ArrayList<>(adapterBeans.values()); AnnotationAwareOrderComparator.sort(this.handlerAdapters); //注入resultHandlers Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors( context, HandlerResultHandler.class, true, false); this.resultHandlers = new ArrayList<>(beans.values()); AnnotationAwareOrderComparator.sort(this.resultHandlers); }
**流程相关源码解析-**HTTPHandlerAutoConfiguration
上面已讲解过WebFlux 核心组件装载过程,那么这些组件又是什么时候注入到对应的容器上下文中的呢?其实是在刷新容器上下文时注入进去的。
org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext#onRefresh
public class ReactiveWebServerApplicationContext extends GenericReactiveWebApplicationContext implements ConfigurableWebServerApplicationContext { @Override protected void onRefresh() { super.onRefresh(); try { createWebServer(); } catch (Throwable ex) { throw new ApplicationContextException("Unable to start reactive web server", ex); } } private void createWebServer() { WebServerManager serverManager = this.serverManager; if (serverManager == null) { String webServerFactoryBeanName = getWebServerFactoryBeanName(); ReactiveWebServerFactory webServerFactory = getWebServerFactory(webServerFactoryBeanName); boolean lazyInit = getBeanFactory().getBeanDefinition(webServerFactoryBeanName).isLazyInit(); // 这里创建容器管理时注入httpHandler this.serverManager = new WebServerManager(this, webServerFactory, this::getHttpHandler, lazyInit); getBeanFactory().registerSingleton("webServerGracefulShutdown", new WebServerGracefulShutdownLifecycle(this.serverManager)); // 注册一个 web容器启动服务类,该类继承了SmartLifecycle getBeanFactory().registerSingleton("webServerStartStop", new WebServerStartStopLifecycle(this.serverManager)); } initPropertySources(); } protected HttpHandler getHttpHandler() { String[] beanNames = getBeanFactory().getBeanNamesForType(HttpHandler.class); if (beanNames.length == 0) { throw new ApplicationContextException( "Unable to start ReactiveWebApplicationContext due to missing HttpHandler bean."); } if (beanNames.length > 1) { throw new ApplicationContextException( "Unable to start ReactiveWebApplicationContext due to multiple HttpHandler beans : " + StringUtils.arrayToCommaDelimitedString(beanNames)); } //容器上下文获取httpHandler return getBeanFactory().getBean(beanNames[0], HttpHandler.class); }
而这个HTTPHandler 是由HTTPHandlerAutoConfiguration装配进去的。
@Configuration @ConditionalOnClass({ DispatcherHandler.class, HttpHandler.class }) @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE) @ConditionalOnMissingBean(HttpHandler.class) @AutoConfigureAfter({ WebFluxAutoConfiguration.class }) @AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE + 10) public class HttpHandlerAutoConfiguration { @Configuration public static class AnnotationConfig { private ApplicationContext applicationContext; public AnnotationConfig(ApplicationContext applicationContext) { this.applicationContext = applicationContext; } //构建WebHandler @Bean public HttpHandler httpHandler() { return WebHttpHandlerBuilder.applicationContext(this.applicationContext) .build(); } }
流程相关源码解析-web容器
org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext#createWebServer 。在创建WebServerManager 容器管理器时会获取对应web容器实例,并注入响应的HTTPHandler。
class WebServerManager { private final ReactiveWebServerApplicationContext applicationContext; private final DelayedInitializationHttpHandler handler; private final WebServer webServer; WebServerManager(ReactiveWebServerApplicationContext applicationContext, ReactiveWebServerFactory factory, Supplier<HttpHandler> handlerSupplier, boolean lazyInit) { this.applicationContext = applicationContext; Assert.notNull(factory, "Factory must not be null"); this.handler = new DelayedInitializationHttpHandler(handlerSupplier, lazyInit); this.webServer = factory.getWebServer(this.handler); } }
以Tomcat 容器为例展示创建过程,使用的是 TomcatHTTPHandlerAdapter 来连接Servlet 请求到HTTPHandler组件。
public class TomcatReactiveWebServerFactory extends AbstractReactiveWebServerFactory implements ConfigurableTomcatWebServerFactory { //隐藏部分代码 @Override public WebServer getWebServer(HttpHandler httpHandler) { if (this.disableMBeanRegistry) { Registry.disableRegistry(); } Tomcat tomcat = new Tomcat(); File baseDir = (this.baseDirectory != null) ? this.baseDirectory : createTempDir("tomcat"); tomcat.setBaseDir(baseDir.getAbsolutePath()); Connector connector = new Connector(this.protocol); connector.setThrowOnFailure(true); tomcat.getService().addConnector(connector); customizeConnector(connector); tomcat.setConnector(connector); tomcat.getHost().setAutoDeploy(false); configureEngine(tomcat.getEngine()); for (Connector additionalConnector : this.additionalTomcatConnectors) { tomcat.getService().addConnector(additionalConnector); } TomcatHttpHandlerAdapter servlet = new TomcatHttpHandlerAdapter(httpHandler); prepareContext(tomcat.getHost(), servlet); return getTomcatWebServer(tomcat); } }
最后Spring容器加载后通过SmartLifecycle实现类WebServerStartStopLifecycle 来启动Web容器。
WebServerStartStopLifecycle 注册过程详见:org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext#createWebServer
3.2 完整请求处理流程
(引用自:https://blog.csdn.net)
该图给出了一个HTTP请求处理的调用链路。是采用Reactor Stream 方式书写,只有最终调用 subscirbe 才真正执行业务逻辑。基于WebFlux 开发时要避免controller 中存在阻塞逻辑。列举下面例子可以看到Spring MVC 和Spring Webflux 之间的请求处理区别。
@RestControllerpublic class TestController { private Logger logger = LoggerFactory.getLogger(this.getClass()); @GetMapping("sync") public String sync() { logger.info("sync method start"); String result = this.execute(); logger.info("sync method end"); return result; } @GetMapping("async/mono") public Mono<String> asyncMono() { logger.info("async method start"); Mono<String> result = Mono.fromSupplier(this::execute); logger.info("async method end"); return result; } private String execute() { try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); } return "hello"; } }
日志输出
2021-05-31 20:14:52.384 INFO 3508 --- [nio-8080-exec-2] c.v.internet.webflux.web.TestController : sync method start 2021-05-31 20:14:57.385 INFO 3508 --- [nio-8080-exec-2] c.v.internet.webflux.web.TestController : sync method end 2021-05-31 20:15:09.659 INFO 3508 --- [nio-8080-exec-3] c.v.internet.webflux.web.TestController : async method start 2021-05-31 20:15:09.660 INFO 3508 --- [nio-8080-exec-3] c.v.internet.webflux.web.TestController : async method end
从上面例子可以看出sync() 方法阻塞了请求,而asyncMono() 没有阻塞请求并立刻返回的。asyncMono() 方法具体业务逻辑 被包裹在了Mono 中Supplier中的了。当execute 处理完业务逻辑后通过回调方式响应给浏览器。
四、存储支持
一旦控制层使用了 Spring Webflux 则安全认证层、数据访问层都必须使用 Reactive API 才真正实现异步非阻塞。
NOSQL Database
-
MongoDB (org.springframework.boot:spring-boot-starter-data-mongodb-reactive)。
-
Redis(org.springframework.boot:spring-boot-starter-data-redis-reactive)。
Relational Database
-
H2 (io.r2dbc:r2dbc-h2)
-
MariaDB (org.mariadb:r2dbc-mariadb)
-
Microsoft SQL Server (io.r2dbc:r2dbc-mssql)
-
MySQL (dev.miku:r2dbc-mysql)
-
jasync-sql MySQL (com.github.jasync-sql:jasync-r2dbc-mysql)
-
Postgres (io.r2dbc:r2dbc-postgresql)
-
Oracle (com.oracle.database.r2dbc:oracle-r2dbc)
五、总结
关于Spring MVC 和Spring WebFlux 测评很多,本文引用下做简单说明。参考:《Spring: Blocking vs non-blocking: R2DBC vs JDBC and WebFlux vs Web MVC》。
基本依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-r2dbc</artifactId> </dependency> <!-- r2dbc 连接池 --> <dependency> <groupId >io.r2dbc</groupId> <artifactId>r2dbc-pool</artifactId> </dependency> <!--r2dbc mysql 库--> <dependency> <groupId>dev.miku</groupId> <artifactId>r2dbc- mysql</artifactId> </dependency> <!--自动配置需要引入一个嵌入式数据库类型对象--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jdbc</artifactId> </dependency> <!-- 反应方程式 web 框架 webflux--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>
相同数据下效果如下**;**
Spring MVC + JDBC 在低并发下表现最好,但 WebFlux + R2DBC 在高并发下每个处理请求使用的内存最少。
Spring WebFlux + R2DBC 在高并发下,吞吐量表现优异。
作者:vivo互联网服务器团队-Zhou Changqing

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
pfSense Software 诞生 15 周年
pfSense Software 官方发文庆祝 pfSense Software 1.0 发布 15 周年,并表示这是一个重要的里程碑。 官方表示,这 15 年里,pfSense Software 达成了许多重大成就,包括: #世界上第一大开源安全网络解决方案 超过 50 个软件版本 超过 26000 次的开放源代码提交 自 2017 年 10 月以来,软件安装量超过 300 万次,遍及每个垂直领域和每个大洲(甚至包括南极洲) 超过338000 名 Netgate 论坛成员 在功能设置、可靠性和易用性方面获得众多赞誉 在这个过程中,Netgate 于 2008 年开始成为 pfSense 项目的主要赞助商,并从 2012 年开始成为该项目的管理人和主要开发者,负责持续的测试、打包、分发、文档等工作。同时,官方对整个 pfSense 社区表示感谢,因为多年来众多才华横溢的忠实开发者做出了大量贡献。 其软件开发总监表示,没有任何一个社区开发的项目同时推动了开源的组件和精神,并将它们结合成一个像 pfSense 软件这样的产品。Netgate 对开源的坚定投资确保数据安全将继续成为所有互联网...
- 下一篇
KuiBaDB —— Rust 重写的 PostgreSQL
KuiBaDB 是使用 Asynchronous Rust 重写的 PostgreSQL,专注于 OLAP 分析。 KuiBaDB 构建在 kbio 和 tokio 之上,只使用 tokio 的'rt-multi-thread', 'rt' 和 'io-util' 特性。所有 IO,包括文件 IO 和网络 IO,以及异步系统调用都由 kbio 提供支持。 KuiBaDB 仅包含 OLAP 数据库所需的基本功能,例如支持事务但不包含子事务。KuiBaDB 使用矢量化引擎,也是目录驱动的。KuiBaDB 使用了 Hologres 中引入的列式存储。不过作者删除了 Delete Map 并为每行添加了 xmin,xmax,xmin/xmax 保存在行存储中。
相关文章
文章评论
共有0条评论来说两句吧...