一、JPA

这是一个悲伤的消息，Spring Data 暂时无法提供响应式的 Spring Data JPA 。这特喵的很严重呀！因为对于我们来说，日常开发最重要的，就是操作 MySQL、Oracle、PostgreSQL、SQLServer 等关系数据库。

这是为什么呢？！

• 问题一，Spring Data JPA 基于 JDBC 实现对数据库的操作，而 JDBC 提供的是阻塞同步的 API 方法。
• 问题二，MySQL、Oracle 等关系数据库是 BIO 模型，一个客户端的 Connection 对应到服务端就是一个线程。这样，就导致 MySQL、Oracle 无法建立大量的客户端连接了。

对于问题一，目前 Oracle 提出了 ADBA(Asynchronous Database Access API) ，而社区提出了 r2dbc(Reactive Relational Database Connectivity) ，希望提供异步非阻塞的 API ，访问数据库。
这样，虽然 MySQL、Oracle 服务器是 BIO 的模型，至少我们在客户端有异步非阻塞的 API 可以调用。想要尝鲜，可以看看 Spring Data R2DBC 项目。 看看就好，目前还处于实验阶段。

对于问题二，需要数据库自身将 BIO 改造成 NIO 的方式，提供支撑大量客户端连接的能力。例如说，国产之光 TiDB 就是基于 NIO 的方式。

当然，还有一种方式，提供数据库 Proxy 代理服务器，提供 NIO 建立连接的方式。这样，即使数据库服务器是 BIO 的方式，Proxy 可以认为是一个数据库的连接池，提供支撑更多的客户端的链接的能力。例如说， Sharding Sphere 提供了 Sharding Proxy ，基于 NIO 的方式实现，可以支持作为 MySQL、PostgreSQL 的 Proxy 服务器。

二、整合响应式的 R2DBC 和事务

我们知道，JDBC 提供的是同步阻塞的数据库访问 API ，那么显然无法在响应式编程中使用，所以 WebFlux 也无法去使用。于是乎 R2DBC 诞生了。
我们可以简单将其理解成响应式的 JDBC 的异步非阻塞 API 。目前其有多种驱动实现，本小节我们会使用 jasync-sql 来访问 MySQL 数据库。
在 Spring Framework 5.2 M2 版本，Spring 提供了 ReactiveTransactionManager 响应式的事务管理器。得益于此，我们可以在响应式变成中，使用 @Transactional 注解来实现声明式事务，又或者使用 TransactionalOperator 来实现编程式事务。本小节，我们会使用 @Transaction 注解来实现声明式事务，毕竟项目中比较少使用编程式事务。

直接使用 JDBC 访问数据库，编写 CRUD 是很繁琐，同理 R2DBC 也是。所以强大的 Spring Data 提供了 Spring Data R2DBC 库，方便我们使用 R2DBC 开发。

下面，让我们来整合 Spring Data R2DBC 到 WebFlux 中，实现响应式的 CRUD 操作。然后实现用户的增删改查接口。接口列表如下：

请求方法	URL	功能
GET	/users/list	查询用户列表
GET	/users/get	获得指定用户编号的用户
POST	/users/add	添加用户
POST	/users/update	更新指定用户编号的用户
POST	/users/delete	删除指定用户编号的用户

2.1 引入依赖

在 [ pom.xml ]文件中，引入相关依赖。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>webflux-r2dbc</artifactId>

    <dependencies>
        <!-- 实现对 Spring WebFlux 的自动化配置 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <!-- 自动化配置响应式的 Spring Data R2DBC -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
        </dependency>

        <!-- jasync 的 r2dbc-mysql 驱动 -->
        <dependency>
            <groupId>com.github.jasync-sql</groupId>
            <artifactId>jasync-r2dbc-mysql</artifactId>
            <version>1.0.11</version>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

    </dependencies>

    <repositories>
        <!-- 引入 Spring 的快照仓库 -->
        <repository>
            <id>spring-libs-snapshot</id>
            <url>https://repo.spring.io/libs-snapshot</url>
        </repository>
        <!-- 引入 Jcenter 的快照仓库 -->
        <repository>
            <id>jcenter</id>
            <url>https://jcenter.bintray.com/</url>
        </repository>
    </repositories>

</project>

2.2 应用配置文件

在 [ application.yml ]中，添加 R2DBC 配置，如下：

spring:
  # R2DBC 配置，对应 R2dbcProperties 配置类
  r2dbc:
    url: mysql://101.133.227.13:3306/orders_1
    username: guo
    password: 205010guo

2.3 DatabaseConfiguration

创建 [DatabaseConfiguration]配置类。代码如下：

package cn.iocoder.springboot.lab27.springwebflux.config;

import com.github.jasync.r2dbc.mysql.JasyncConnectionFactory;
import com.github.jasync.sql.db.mysql.pool.MySQLConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;
import org.springframework.boot.autoconfigure.r2dbc.R2dbcProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.connectionfactory.R2dbcTransactionManager;
import org.springframework.transaction.ReactiveTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

import java.net.URI;
import java.net.URISyntaxException;

@Configuration
@EnableTransactionManagement // 开启事务的支持
public class DatabaseConfiguration {

    @Bean
    public ConnectionFactory connectionFactory(R2dbcProperties properties) throws URISyntaxException {
        // 从 R2dbcProperties 中，解析出 host、port、database
        URI uri = new URI(properties.getUrl());
        String host = uri.getHost();
        int port = uri.getPort();
        String database = uri.getPath().substring(1); // 去掉首位的 / 斜杠
        // 创建 jasync Configuration 配置配置对象
        com.github.jasync.sql.db.Configuration configuration = new com.github.jasync.sql.db.Configuration(
                properties.getUsername(), host, port, properties.getPassword(), database);
        // 创建 JasyncConnectionFactory 对象
        return  new JasyncConnectionFactory(new MySQLConnectionFactory(configuration));
    }

    @Bean
    public ReactiveTransactionManager transactionManager(R2dbcProperties properties) throws URISyntaxException {
        return new R2dbcTransactionManager(this.connectionFactory(properties));
    }

}

• 通过 @EnableTransactionManagement 注解，开启 Spring Transaction 的支持。
• #connectionFactory(properties) 方法，创建 JasyncConnectionFactory Bean 对象。因为 spring-boot-starter-data-r2dbc 支持 R2DBC 的自动化配置，但是暂不支持自动创建 JasyncConnectionFactory 作为 ConnectionFactory Bean ，所以这里我们需要自定义。
• #transactionManager(properties) 方法，创建响应式的 R2dbcTransactionManager 事务管理器。

2.4 Application

创建 [ Application.java ]类，配置 @SpringBootApplication 注解即可。代码如下：

package cn.iocoder.springboot.lab27.springwebflux;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

2.5 UserRepository

package cn.iocoder.springboot.lab27.springwebflux.dao;

import cn.iocoder.springboot.lab27.springwebflux.dataobject.UserDO;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Mono;

public interface UserRepository extends ReactiveCrudRepository<UserDO, Integer> {

    @Query("SELECT id, username, password, create_time FROM users WHERE username = :username")
    Mono<UserDO> findByUsername(String username);

}

• 对应的实体为 UserDO.java 类。对应建表语句见 users.sql 文件。

• 实现 ReactiveCrudRepository 接口，它是响应式的 Repository 基础接口。

• #findByUsername(String username) 方法，定义了查询指定用户名的用户，返回的结果也算是使用 Mono 包装过。

  • 注意，这里的 @Query 注解是 Spring Data R2DBC 自定义的，而不是 JPA 规范中的。
  • 如果我们注释掉 @Query 注解，启动项目会报 “Query derivation not yet supported!” 异常提示。目前看下来，Spring Data R2DBC 暂时不支持【基于方法名查询】。

2.6 UserController

package cn.iocoder.springboot.lab27.springwebflux.controller;

import cn.iocoder.springboot.lab27.springwebflux.dao.UserRepository;
import cn.iocoder.springboot.lab27.springwebflux.dataobject.UserDO;
import cn.iocoder.springboot.lab27.springwebflux.dto.UserAddDTO;
import cn.iocoder.springboot.lab27.springwebflux.dto.UserUpdateDTO;
import cn.iocoder.springboot.lab27.springwebflux.vo.UserVO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Date;
import java.util.Objects;
import java.util.function.Function;

/**
 * 用户 Controller
 */
@RestController
@RequestMapping("/users")
public class UserController {

    private static final UserDO USER_NULL = new UserDO();

    @Autowired
    private UserRepository userRepository;

    /**
     * 查询用户列表
     *
     * @return 用户列表
     */
    @GetMapping("/list")
    public Flux<UserVO> list() {
        // 返回列表
        return userRepository.findAll()
                .map(userDO -> new UserVO().setId(userDO.getId()).setUsername(userDO.getUsername()));
    }

    /**
     * 获得指定用户编号的用户
     *
     * @param id 用户编号
     * @return 用户
     */
    @GetMapping("/get")
    public Mono<UserVO> get(@RequestParam("id") Integer id) {
        // 返回
        return userRepository.findById(id)
                .map(userDO -> new UserVO().setId(userDO.getId()).setUsername(userDO.getUsername()));
    }

    /**
     * 添加用户
     *
     * @param addDTO 添加用户信息 DTO
     * @return 添加成功的用户编号
     */
    @PostMapping("add")
    @Transactional
    public Mono<Integer> add(UserAddDTO addDTO) {
        // 查询用户
        Mono<UserDO> user = userRepository.findByUsername(addDTO.getUsername());

        // 执行插入
        return user.defaultIfEmpty(USER_NULL) // 设置 USER_NULL 作为 null 的情况，否则 flatMap 不会往下走
                .flatMap(new Function<UserDO, Mono<Integer>>() {

                    @Override
                    public Mono<Integer> apply(UserDO userDO) {
                        if (userDO != USER_NULL) {
                            // 返回 -1 表示插入失败。
                            // 实际上，一般是抛出 ServiceException 异常。因为这个示例项目里暂时没做全局异常的定义，所以暂时返回 -1 啦
                            return Mono.just(-1);
                        }
                        // 将 addDTO 转成 UserDO
                        userDO = new UserDO()
                                .setUsername(addDTO.getUsername())
                                .setPassword(addDTO.getPassword())
                                .setCreateTime(new Date());
                        // 插入数据库
                        return userRepository.save(userDO).flatMap(new Function<UserDO, Mono<Integer>>() {
                            @Override
                            public Mono<Integer> apply(UserDO userDO) {
                                // 如果编号为偶数，抛出异常。
                                if (userDO.getId() % 2 == 0) {
                                    throw new RuntimeException("我就是故意抛出一个异常，测试下事务回滚");
                                }

                                // 返回编号
                                return Mono.just(userDO.getId());
                            }
                        });
                    }

                });
    }

    /**
     * 更新指定用户编号的用户
     *
     * @param updateDTO 更新用户信息 DTO
     * @return 是否修改成功
     */
    @PostMapping("/update")
    public Mono<Boolean> update(UserUpdateDTO updateDTO) {
        // 查询用户
        Mono<UserDO> user = userRepository.findById(updateDTO.getId());

        // 执行更新
        return user.defaultIfEmpty(USER_NULL) // 设置 USER_NULL 作为 null 的情况，否则 flatMap 不会往下走
                .flatMap(new Function<UserDO, Mono<Boolean>>() {

                    @Override
                    public Mono<Boolean> apply(UserDO userDO) {
                        // 如果不存在该用户，则直接返回 false 失败
                        if (userDO == USER_NULL) {
                            return Mono.just(false);
                        }
                        // 查询用户是否存在
                        return userRepository.findByUsername(updateDTO.getUsername())
                                .defaultIfEmpty(USER_NULL) // 设置 USER_NULL 作为 null 的情况，否则 flatMap 不会往下走
                                .flatMap(new Function<UserDO, Mono<? extends Boolean>>() {

                                    @Override
                                    public Mono<? extends Boolean> apply(UserDO usernameUserDO) {
                                        // 如果用户名已经使用（该用户名对应的 id 不是自己，说明就已经被使用了）
                                        if (usernameUserDO != USER_NULL && !Objects.equals(updateDTO.getId(), usernameUserDO.getId())) {
                                            return Mono.just(false);
                                        }
                                        // 执行更新
                                        userDO.setUsername(updateDTO.getUsername());
                                        userDO.setPassword(updateDTO.getPassword());
                                        return userRepository.save(userDO).map(userDO -> true); // 返回 true 成功
                                    }

                                });
                    }

                });
    }

    /**
     * 删除指定用户编号的用户
     *
     * @param id 用户编号
     * @return 是否删除成功
     */
    @PostMapping("/delete") // URL 修改成 /delete ，RequestMethod 改成 DELETE
    public Mono<Boolean> delete(@RequestParam("id") Integer id) {
        // 查询用户
        Mono<UserDO> user = userRepository.findById(id);

        // 执行删除。这里仅仅是示例，项目中不要物理删除，而是标记删除
        return user.defaultIfEmpty(USER_NULL) // 设置 USER_NULL 作为 null 的情况，否则 flatMap 不会往下走
                .flatMap(new Function<UserDO, Mono<Boolean>>() {

                    @Override
                    public Mono<Boolean> apply(UserDO userDO) {
                        // 如果不存在该用户，则直接返回 false 失败
                        if (userDO == USER_NULL) {
                            return Mono.just(false);
                        }
                        // 执行删除
                        return userRepository.deleteById(id).map(aVoid -> true); // 返回 true 成功
                    }

                });
    }

}

注意，我们在 #add(UserAddDTO addDTO) 方法上，添加了 @Transactional 注解，表示整个逻辑需要在事务中进行。同时，为了模拟事务回滚的情况，我们故意在插入记录的 id 编号为偶数时，抛出 RuntimeException 异常，回滚事务。

2.7 测试

访问url： http://localhost:8080/users/list 。
返回：

[
    {
        "id": 1,
        "username": "35e89c7b-bd4b-45d1-98d0-567db0181b48"
    },
    {
        "id": 5,
        "username": "26d29bbd-d38f-4ae1-8ff8-bfbf6afc23bc"
    },
    {
        "id": 6,
        "username": "1b20f566-2084-41e7-84e1-68fb37463362"
    },
    {
        "id": 7,
        "username": "26446564-9c8c-4bd7-98e1-f0102c2713b3"
    },
    {
        "id": 8,
        "username": "60a0357b-b591-4406-9dfe-4710022a4851"
    },
    {
        "id": 9,
        "username": "bfcbba14-770a-4918-8786-9013ca84664b"
    }
]

底线

---

本文源代码使用 Apache License 2.0 开源许可协议， 这里是本文源码Gitee地址 ，可通过命令 git clone+地址 下载代码到本地，也可直接点击链接通过浏览器方式查看源代码。