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: 9999
servlet:
context-path: /
spring:
#连接数据库的url,前缀不再是jdbc而是换成r2dbc
#这里可以配置连接池相关的其它属性,这里为了简洁不配置
r2dbc:
url: r2dbc:mysql://localhost:3306/tope-pay-user?&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&useSSL=false
username: root
password: 123456
logging:
level:
org.springframework.r2dbc: INFO #输出执行的sql
org.springframework.cloud.web.reactive: info
reactor.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 { @Transient private 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; /** * 系统包含那些模块 * 该字段不参与数据库映射 */ @Transient private 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 进行数据自动填充。 */ @Version private Integer version; }
业务接口与实现层:
package org.jd.websocket.auth.data.reactor.service; import org.jd.websocket.auth.data.reactor.entity.RSysSystem; import reactor.core.publisher.Mono; 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(); }接口实现层:
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.Mono; import javax.annotation.Resource; /** * 构建全调用链路异步响应式编程 * 系统响应式查询服务 */ @Slf4j @Service public class RSysSystemServiceImpl implements RSysSystemService { @Resource private RSysSystemReactiveRepository sysSystemReactiveRepository; @Override public Mono<RSysSystem> findById(Long systemId) { return sysSystemReactiveRepository.findById(systemId); } @Override @Transactional(rollbackFor = Exception.class) public Mono<RSysSystem> insert(RSysSystem system) { return sysSystemReactiveRepository.save(system); } @Override public Mono<Boolean> exists(Long systemId) { return sysSystemReactiveRepository.existsById(systemId); } @Override public Mono<Long> count() { return sysSystemReactiveRepository.count(); } }
暴露服务端点:
package org.jd.websocket.auth.data.reactor.controller; import org.jd.websocket.auth.data.reactor.entity.RSysSystem; import org.jd.websocket.auth.data.reactor.service.RSysSystemService; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; import javax.annotation.Resource; @RestController @RequestMapping("/system") public class SysSystemController { @Resource private RSysSystemService rSysSystemService; @GetMapping("/getSysSystemById/{systemId}") public Mono<RSysSystem> getSySystem(@PathVariable("systemId") Long systemId) { Mono<RSysSystem> result = rSysSystemService.findById(systemId); System.out.println("result:" + result.toString()); return result; } }
访问测试:
http://localhost:9999/system/getSysSystemById/1
结果:
至此,基础搭建完成,后续会持续系列多篇讲解,撸下源代码及相关知识........待续.....
参考序列:
* 官方文档
* https://github.com/spring-projects/spring-data-examples/tree/master/r2dbc/example
* https://www.reactiveprinciples.org/ 中文官网
https://github.com/jasync-sql/jasync-sql 参考官网还是老外有耐心........