(二)Spring WeFlux响应式编程第二种整合方案|道法术器

本文主要是介绍(二)Spring WeFlux响应式编程第二种整合方案|道法术器，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

特注意: Spring Boot 的版本大于等于 2.3.0，在此版本之后才开始支持 MYSQL 的响应式驱动

在执行程序时: 通常为了提供性能,处理器和编译器常常会对指令进行重排序。从排序分为编译器重排序和处理器重排序两种：

(1)编译器重排序: 编译器保证不改变单线程执行结构的前提下,可以调整多线程语句执行顺序； (2)处理器重排序：如果不存在数据依赖性,处理器可以改变语句对应机器指令的执行顺序;

若要实现快速响应,就得把程序执行指令的方式换一换,将同步方式改成异步方法,方法执行改为消息发送,因此诞生了响应式编程模型；

特注意的是: 在响应式编程技术栈中,有一点需要注意,即响应式编程并不是针对系统中的某一部分组件,而是需要使用于"调用链路"上的所有组件。无论是Web层(Controller), 服务层(Service) ,还是处于下游的数据访问层(Dao或者Repository), 只要有一个环节不是响应式的, 那么这个环节就会出现“同步阻塞”，从而导致"背压机制"失效 , 这就是所谓的全栈式响应编程的设计理念。

Spring WebFlux 响应式异步编程|道法术器(一)

Spring WeFlux响应式编程整合另一种方案|道法术器(二)

R2dbc操作mysql 注意下面红色部分与上一篇"Spring WebFlux 响应式异步编程|道法术器(一)" 不一样的依赖包

技术整合:

<!--设置spring-boot依赖的版本 -->
<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.5</version> <!--2.4.11--><relativePath/> <!-- lookup parent from repository -->
</parent>

<!--R2DBC是基于Reactive Streams标准来设计的。通过使用R2DBC，你可以使用reactive API来操作数据。同时R2DBC只是一个开放的标准，而各个具体的数据库连接实现，需要实现这个标准。-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<!--响应式编程传统的jdbc操作是阻塞式的，所以不能再用以前的mysql驱动了-->
<dependency><groupId>dev.miku</groupId><artifactId>r2dbc-mysql</artifactId>
</dependency>

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope>
</dependency>
<!--reactor-test测试相关类-->
<dependency><groupId>io.projectreactor</groupId><artifactId>reactor-test</artifactId>
</dependency>

<!-- 响应式编程集成-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId>
</dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><scope>provided</scope>
</dependency>
<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId>
</dependency><!--连接mysql数据库的驱动 -->
<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId>
</dependency>

application.yml配置文件: 配置文件url与上一篇有细小的区别, 不小心就驱动保错:

报错信息:

     at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:791) ~[spring-beans-5.3.17.jar:5.3.17]
   ... 63 common frames omitted
Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'connectionFactory' defined in class path resource [org/springframework/boot/autoconfigure/r2dbc/ConnectionFactoryConfigurations$PoolConfiguration$PooledConnectionFactoryConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [io.r2dbc.pool.ConnectionPool]: Factory method 'connectionFactory' threw exception; nested exception is java.lang.IllegalArgumentException: URL mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=false does not start with the r2dbc scheme
   at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:658) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:638) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1352) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1195) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:582) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:542) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:335) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:333) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:208) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1389) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1309) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:887) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:791) ~[spring-beans-5.3.17.jar:5.3.17]
   ... 77 common frames omitted
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [io.r2dbc.pool.ConnectionPool]: Factory method 'connectionFactory' threw exception; nested exception is java.lang.IllegalArgumentException: URL mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=false does not start with the r2dbc scheme
   at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.3.17.jar:5.3.17]
   at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:653) ~[spring-beans-5.3.17.jar:5.3.17]
   ... 91 common frames omitted
Caused by: java.lang.IllegalArgumentException: URL mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=false does not start with the r2dbc scheme
   at io.r2dbc.spi.ConnectionUrlParser.validate(ConnectionUrlParser.java:54) ~[r2dbc-spi-0.8.6.RELEASE.jar:na]
   at io.r2dbc.spi.ConnectionUrlParser.parseQuery(ConnectionUrlParser.java:81) ~[r2dbc-spi-0.8.6.RELEASE.jar:na]
   at io.r2dbc.spi.ConnectionFactoryOptions.parse(ConnectionFactoryOptions.java:122) ~[r2dbc-spi-0.8.6.RELEASE.jar:na]
   at org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryOptionsInitializer.initializeRegularOptions(ConnectionFactoryOptionsInitializer.java:60) ~[spring-boot-autoconfigure-2.6.5.jar:2.6.5]
   at org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryOptionsInitializer.initialize(ConnectionFactoryOptionsInitializer.java:49) ~[spring-boot-autoconfigure-2.6.5.jar:2.6.5]
   at org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryConfigurations.createConnectionFactory(ConnectionFactoryConfigurations.java:62) ~[spring-boot-autoconfigure-2.6.5.jar:2.6.5]
   at org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryConfigurations$PoolConfiguration$PooledConnectionFactoryConfiguration.connectionFactory(ConnectionFactoryConfigurations.java:92) ~[spring-boot-autoconfigure-2.6.5.jar:2.6.5]
   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_221]
   at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_221]
   at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_221]
   at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_221]
   at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154) ~[spring-beans-5.3.17.jar:5.3.17]
   ... 92 common frames omitted

大致的意思：不能根据配置文件初始化R2dbc数据库连接工厂:

sjava.lang.IllegalArgumentException: URL mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=false

server:port: 9999servlet:context-path: /
spring:#连接数据库的url，前缀不再是jdbc而是换成r2dbc#这里可以配置连接池相关的其它属性，这里为了简洁不配置r2dbc:url: r2dbc:mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=falseusername: rootpassword: 123456logging:level:org.springframework.r2dbc: INFO  #输出执行的sqlorg.springframework.cloud.web.reactive: inforeactor.ipc.netty: info

数据持久层:

package org.jd.websocket.auth.data.reactor.repository;import org.jd.websocket.auth.data.reactor.entity.RSysSystem;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.data.repository.reactive.ReactiveSortingRepository;
import org.springframework.stereotype.Repository;/*** 系统服务类异步编程模型接口* 增强了排序功能*/
@Repository
public interface RSysSystemReactiveRepository extends ReactiveCrudRepository<RSysSystem, Long>, ReactiveSortingRepository<RSysSystem, Long> {
}

业务领域模型:

package org.jd.websocket.auth.data.reactor.entity;import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Transient;
import org.springframework.data.annotation.Version;
import org.springframework.data.relational.core.mapping.Column;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.format.annotation.DateTimeFormat;/*** 属性上的注解使用Spring-data中的相关注解*/
import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.List;@Data
@RequiredArgsConstructor
@Table(value = "sys_system")
public class RSysSystem implements Serializable {@Transientprivate static final long serialVersionUID = 7481799808203597699L;// 主键自增@Id@Column(value = "system_id")private Long systemId;/*** 系统名称* 字段映射和约束条件* //对应数据库表中哪个列字段及对该字段的自定义*/@Column(value = "system_name")private String systemName;/*** 详细功能描述: 描述该系统主要包含那些那些模块，每个模块的大致功能*/@Column(value = "system_detail_desc")private String systemDetailDesc;/*** 系统跳转到功能版块路径*/@Column(value = "path_function_url")private String pathFunctionUrl;/*** 系统包含那些模块* 该字段不参与数据库映射*/@Transientprivate List<RSysModule> sysModules;/**** 创建时间*/@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")@Column(value = "create_time")private LocalDateTime createTime;/*** 更新时间*/@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")@Column(value = "update_time")private LocalDateTime updateTime;/*** 版本号（用于乐观锁, 默认为 1）* 使用 @Version 注解标注对应的实体类。* 可以通过 @TableField 进行数据自动填充。*/@Versionprivate Integer version;
}

业务接口与实现层:

package org.jd.websocket.auth.data.reactor.service;import org.jd.websocket.auth.data.reactor.entity.RSysSystem;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import java.util.List;public interface RSysSystemService {/*** 通过ID查找单条记录** @param systemId 系统服务ID* @return {@link Mono<RSysSystem>}*/Mono<RSysSystem> findById(Long systemId);/*** 插入记录信息** @param system* @return {@link Mono<RSysSystem>)*/Mono<RSysSystem> insert(RSysSystem system);/*** 通过ID查询是否存在记录** @param systemId 系统ID* @return {@link Mono<Boolean>}*/Mono<Boolean> exists(Long systemId);/*** 查询记录数** @return {@link Mono<Long>}*/Mono<Long> count();/*** 查询所有* @return {@link Flux<RSysSystem>}*/Flux<RSysSystem> findAll();
}

接口实现层:

package org.jd.websocket.auth.data.reactor.service.impl;import lombok.extern.slf4j.Slf4j;
import org.jd.websocket.auth.data.reactor.entity.RSysSystem;
import org.jd.websocket.auth.data.reactor.repository.RSysSystemReactiveRepository;
import org.jd.websocket.auth.data.reactor.service.RSysSystemService;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import javax.annotation.Resource;
import java.util.List;/*** 构建全调用链路异步响应式编程* 系统响应式查询服务*/
@Slf4j
@Service
public class RSysSystemServiceImpl implements RSysSystemService {@Resourceprivate R2dbcEntityTemplate r2dbcEntityTemplate;@Resourceprivate RSysSystemReactiveRepository sysSystemReactiveRepository;@Overridepublic Mono<RSysSystem> findById(Long systemId) {return sysSystemReactiveRepository.findById(systemId);}@Override@Transactional(rollbackFor = Exception.class)public Mono<RSysSystem> insert(RSysSystem system) {return sysSystemReactiveRepository.save(system);}@Overridepublic Mono<Boolean> exists(Long systemId) {return sysSystemReactiveRepository.existsById(systemId);}@Overridepublic Mono<Long> count() {return sysSystemReactiveRepository.count();}@Overridepublic Flux<RSysSystem> findAll() {return sysSystemReactiveRepository.findAll();}
}

暴露服务端点: 全调用链路都采用异步编程思想

写一个帮助类:

package org.jd.websocket.auth.data.reactor.help;import org.jd.websocket.auth.data.reactor.entity.RSysSystem;
import org.jd.websocket.auth.data.reactor.service.RSysSystemService;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import javax.annotation.Resource;/*** 创建一个辅助类来调用Service*/
@Component
public class SysSystemHandler {@Resourceprivate RSysSystemService rSysSystemService;public Mono<ServerResponse> queryById(ServerRequest serverRequest) {// 获取URL上面携带的参数systemId;Long systemId = Long.valueOf(serverRequest.pathVariable("systemId"));Mono<RSysSystem> sysSystemMono = rSysSystemService.findById(systemId);return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(sysSystemMono, RSysSystem.class);}/*** 查询所有** @param serverRequest* @return {@link Mono<ServerResponse>}*/public Mono<ServerResponse> queryAllList(ServerRequest serverRequest) {Flux<RSysSystem> sysSystemFlux = rSysSystemService.findAll();return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(sysSystemFlux, RSysSystem.class);}
}

来一个配置类:

(1)功能相当于@RequestMapping，在WebFlux中,采用函数式开发模式中,提供类似HandlerFunction和RouterFunction接口来实现Spring MVC的类似功能;

(2) HandlerFunction相当于Controller中具体处理方法,输入为请求，输出为封装在Mono中的响应；(3)RouterFunction相当于@RequestMapping，将URL映射到具体的HandlerFunction输入请求,输出封装在Mono中的HandlerFunction。

编写一个配置类:

package org.jd.websocket.auth.data.reactor.controller.config;import org.jd.websocket.auth.data.reactor.help.SysSystemHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;/*** 创建一个路由配置类：RouterFunction*/
@Configuration
public class SysSystemRoutingConfig {@Beanpublic RouterFunction<ServerResponse> monoRouterFunction(SysSystemHandler sysSystemHandler) {return route(GET("/system/getSysSystemById/{systemId}").and(accept(MediaType.APPLICATION_JSON)), sysSystemHandler::queryById).andRoute(GET("/system/all").and(accept(MediaType.APPLICATION_JSON)),sysSystemHandler::queryAllList);}}

访问测试:

http://localhost:9999/system/getSysSystemById/1 获取id=1的记录

http://localhost:9999/system/all 获取所有记录

结果: