Spring Data R2DBC响应式操作MySQL

1. 前言

在 使用R2DBC操作 MySQL 数据库 一文中初步介绍了 r2dbc-mysql 的使用。由于借助 DatabaseClient 操作 MySQL ，过于初级和底层，不利于开发。今天就利用 Spring Data R2DBC 来演示 Spring 数据存储抽象（Spring Data Repository） 风格的 R2DBC 数据库操作。

请注意 ：目前 Spring Data R2DBC 虽然已经迭代了多个正式版，但是仍然处于初级阶段，还不足以运用到生产中。不过未来可期，值得研究学习。

2. Spring Data R2DBC

Spring Data R2DBC提供了基于 R2DBC 反应式关系数据库驱动程序的流行的 Repository 抽象。但是这并不是一个ORM框架，你可以把它看做一个数据库访问的抽象层或者 R2DBC 的客户端程序。它不提供 ORM 框架具有的缓存、懒加载等诸多特性，但它抽象了数据库和对象的抽象映射关系，具有轻量级、易用性的特点。

2.1 版本对应关系

胖哥总结了截至目前 Spring Data R2DBC 和 Spring Framework 的版本对应关系：

一定要注意版本对应关系，避免不兼容的情况。

3. 基础依赖

上次我没有引用 R2DBC 连接池，这次我将尝试使用它。主要依赖如下 ，这里我还集成了 Spring Webflux :

<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 Boot 2.3.2.RELEASE 。

4. 配置

上次我们采用的是 JavaConfig 风格的配置，只需要向 Spring IoC 注入一个 ConnectionFactory 。这一次我将尝试在 application.yaml 中配置 R2DBC 的必要参数。

spring:
  r2dbc:
    url: r2dbcs:mysql://127.0.0.1:3306/r2dbc
    username: root
    password: 123456

以上就是 R2DBC 的主要配置。特别注意的是 spring.r2dbc.url 的格式，根据数据库的不同写法是不同的，要看驱动的定义，这一点非常重要。连接池这里使用默认配置即可，不用显式定义。

5. 编写业务代码

接下来就是编写业务代码了。这里我还尝试使用 DatabaseClient 来执行了 DDL 语句创建了 client_user 表，感觉还不错。

@Autowired
DatabaseClient databaseClient;

@Test
void doDDL() {

    List<String> ddl = Collections.unmodifiableList(Arrays.asList("drop table if exists client_user;", "create table client_user(user_id varchar(64) not null primary key,nick_name varchar(32),phone_number varchar(16),gender tinyint default 0) charset = utf8mb4;"));
    ddl.forEach(sql -> databaseClient.execute(sql)
            .fetch()
            .rowsUpdated()
            .as(StepVerifier::create)
            .expectNextCount(1)
            .verifyComplete());
}

5.1 声明数据库实体

熟悉 Spring Data JPA 的同学应该很轻车熟路了。

/**
 *  the client user type
 *
 * @author felord.cn
 */
@Data
@Table
public class ClientUser implements Serializable {
    private static final long serialVersionUID = -558043294043707772L;
    @Id
    private String userId;
    private String nickName;
    private String phoneNumber;
    private Integer gender;
}

5.2 声明CRUD接口

上面实体类中的 @Table 注解是有说法的，当我们的操作接口继承的是 ReactiveCrudRepository<T, ID> 或者 ReactiveSortingRepository<T, ID> 时，需要在实体类上使用 @Table 注解，这也是推荐的用法。

public interface ReactiveClientUserSortingRepository extends ReactiveSortingRepository<ClientUser,String> {
    
}

当然实体类不使用 @Table 注解标记时，我们还可以继承 R2dbcRepository<T, ID> 接口。然后 ReactiveClientUserSortingRepository 将提供一些操作数据库的方法。

然后 Spring Data JPA 怎么写，这里也差不多怎么写，但是有些功能现在还没有得到支持，比如上面提到的分页，还有主键策略等。

类似 PagingAndSortingRepository<T,ID> 的反应式分页功能接口目前还没有实装，会在未来的版本集成进来。

5.3 实际操作

接下来我们就要通过 R2DBC 实际操作 MySQL 数据库了。按照我们传统的逻辑写了如下的新增逻辑：

ClientUser clientUser = new ClientUser();

clientUser.setGender(2);
clientUser.setNickName("r2dbc");
clientUser.setPhoneNumber("9527");
clientUser.setUserId("snowflake");

Mono<ClientUser> save = reactiveClientUserSortingRepository.save(clientUser);

结果数据库并没有写入数据。这时因为 r2dbc-mysql 不能被直接使用，只能由客户端去实现并委托给客户端去操作。

这也是 R2DBC 的设计原则，R2DBC的目标是最小化SPI平面，目的是消除数据库之间的差异部分，并使得整个数据库完全具有反应式和背压。它主要用作客户端库使用的驱动程序SPI，而不打算直接在应用程序代码中使用。

所以这里我们可以借助于 reactor-test 测试库去执行一下，改写为：

reactiveClientUserSortingRepository.save(clientUser)
        .log()
        .as(StepVerifier::create)
        .expectNextCount(1)
        .verifyComplete();

但是依然不能执行成功，提示 update table [client_user]. Row with Id [snowflake] does not exist ，也就是说期望执行的是新增但是实际执行的是更新，由于数据库找不到主键为 snowflake 的记录就报了错。这里为什么是更新呢？

这时因为实体类在进行新增时会判断主键是否填充，如果没有填充就认为是新数据，采取真正的新增操作，主键需要数据库来自动填充；如果主键存在值则认为是旧数据则调用更新操作。胖哥同 Spring Data R2DBC 的项目组沟通后并没有得到友好的解决方案，不过我已经找到了方法，这里先留个坑。

那么该如何新增一条数据呢？我们只能借助于 @Query 注解来编写一条 SQL 写入了：

@Modifying
@Query("insert into client_user (user_id,nick_name,phone_number,gender) values (:userId,:nickName,:phoneNumber,:gender)")
Mono<Integer> addClientUser(String userId, String nickName, String phoneNumber, Integer gender);

当添加了 @Modifying 后，返回值可以从 Mono<ClientUser> 、 Mono<Boolean> 或者 Mono<Integer> 任意一种选择。

reactiveClientUserSortingRepository
        .addClientUser("snowflake",
                "r2dbc",
                "132****155",
                0)
        .as(StepVerifier::create)
        .expectNextCount(1)
        .verifyComplete();

这样就证明写成功了一条数据。

5.4 搭配Webflux使用

但是实际中该如何应用呢？目前能够想到的就是结合反应式框架 Spring Webflux 了，就像 Spring Data JPA 配合 Spring MVC 一样。

我们编写一个 Webflux 接口:

@RestController
@RequestMapping("/user")
public class ReactiveClientUserController {

    @Autowired
    private ReactiveClientUserSortingRepository reactiveClientUserSortingRepository;


    /**
     * 这里为了检验默认api 就不分层了
     *
     * @param userId the user id
     * @return the mono
     */
    @GetMapping("/{userId}")
    public Mono<ClientUser> findUserById(@PathVariable String userId) {
        return reactiveClientUserSortingRepository.findById(userId);
    }

}

5.5 一些测试数据参考

在低并发时， Spring MVC + JDBC 表现最佳，但在高并发下， WebFlux + R2DBC 使用每个已处理请求的内存最少。

在高并发下， Spring MVC + JDBC 的响应时间开始下降。显然， R2DBC 在更高的并发性下提供了更好的响应时间。 Spring WebFlux 也比使用 Spring MVC 的类似实现更好。

6. 总结

今天对 Spring Data R2DBC 进一步演示，相信你能够从中学到一些东西。由于 R2DBC 还是比较新，还存在一些需要改进和补充的东西。目前社区非常活跃，发展十分迅速。好了今天的文章就到这里，原创不易多多关注： 码农小胖哥 如果你觉得本文很有用，请点赞、转发、再看。

关注公众号：Felordcn 获取更多资讯

个人博客：https://felord.cn