少年辛苦终身事,莫向光阴惰寸功。这篇文章主要讲述深入剖析 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容器中。
- 适用高并发
- 高吞吐量
- 可伸缩性
文章图片
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/jsonWebFlux 默认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.384INFO 3508 --- [nio-8080-exec-2] c.v.internet.webflux.web.TestController: sync method start
2021-05-31 20:14:57.385INFO 3508 --- [nio-8080-exec-2] c.v.internet.webflux.web.TestController: sync method end
2021-05-31 20:15:09.659INFO 3508 --- [nio-8080-exec-3] c.v.internet.webflux.web.TestController: async method start
2021-05-31 20:15:09.660INFO 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)。
- 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)
基本依赖
<
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】Spring WebFlux + R2DBC 在高并发下,吞吐量表现优异。
推荐阅读
- 首发Citrix/view虚拟桌面禁用普通用户更改或删除网卡权限
- 信息流推荐系统智能交付解决方案探索
- 计算机网络基础
- 10.4-10.10博客精彩回顾
- 递归问题闲究——青蛙跳台阶
- 以小窥大,从一盏路灯看亿万物联网之路
- 智汀家庭云-快速入门(使用Docker运行)
- 实体链接在OPPO小布助手和OGraph的实践应用
- uni-app技术分享| 用uni-app实现拖动的诀窍