java - 为什么注释 @Transactional 在自定义 AbstractR2dbcConfiguration 的情况下不起作用

Question

最近在学习r2dbc，遇到一个问题。为了测试 r2dbc 中的事务，我写了一个小测试项目。 sample code github

在ma​​ster分支可以看到正确的代码，在zd/transactional-test分支可以看到错误的代码

首先我们有一个mysql表是这样的:

CREATE TABLE `users` (
     `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
     `username` varchar(64) COLLATE utf8mb4_bin DEFAULT NULL COMMENT 'username',
     `password` varchar(32) COLLATE utf8mb4_bin DEFAULT NULL COMMENT 'password',
     `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT 'crete time',
     PRIMARY KEY (`id`),
     UNIQUE KEY `idx_username` (`username`)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

当代码如下时，注解@Transactionl效果很好

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">
    <modelVersion>4.0.0</modelVersion>
    <groupId>org.example</groupId>
    <artifactId>webflux-transactional</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>14</maven.compiler.source>
        <maven.compiler.target>14</maven.compiler.target>
    </properties>


    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
            <version>2.4.4</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
            <version>2.4.4</version>
        </dependency>

        <dependency>
            <groupId>io.r2dbc</groupId>
            <artifactId>r2dbc-pool</artifactId>
            <version>0.8.6.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jdbc</artifactId>
            <version>5.3.5</version>
        </dependency>

        <dependency>
            <groupId>com.github.jasync-sql</groupId>
            <artifactId>jasync-r2dbc-mysql</artifactId>
            <version>1.1.6</version>
            <exclusions>
                <exclusion>
                    <groupId>io.netty</groupId>
                    <artifactId>netty-transport</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>io.netty</groupId>
                    <artifactId>netty-handler</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>io.projectreactor</groupId>
                    <artifactId>reactor-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <!--test-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <version>2.4.4</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>3.4.4</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <version>3.4.4</version>
            <scope>test</scope>
        </dependency>

    </dependencies>

</project>

1. 应用程序.yaml

spring:
  r2dbc:
    url: r2dbc:pool:mysql://127.0.0.1:3306/webflux-r2dbc  # r2dbc:mysql://127.0.0.1:3306/demo
    username: root
    password: 123456
    pool:
      initial-size: 5
      max-size: 500
      max-idle-time: 30m
      validation-query: SELECT 1

server:
  port: 8081

1. 演示应用

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

1. 实体类

@Table(value = "users")
public class User {
    @Id
    private Integer id;
    private String username;
    private String password;
//omit getter setter
}

1. 用户资料库:

public interface UserRepository extends ReactiveCrudRepository<User, Integer> {

}

1. 用户服务

@Service
public class UserService {

    private final static Logger LOGGER = LoggerFactory.getLogger(UserService.class);

    @Resource
    R2dbcEntityTemplate r2dbcEntityTemplate;

    @Transactional(rollbackFor = Exception.class)
    public Mono<Integer> add1(User queryUser) {

        return this.r2dbcEntityTemplate.insert(User.class)
                .using(queryUser)
                .doOnSuccess(user -> {
                    if (!user.getUsername().contains("exception")) {
                        LOGGER.info("=====================add normal=================");
                    } else {
                        LOGGER.error("=====================add exception=================");
                        throw new RuntimeException("add1 exception test............");
                    }

                })
                .map(User::getId);
    }
}

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

    @Autowired
    private UserService userService;

    /**
     * test Transactional
     * @param user
     * @return
     */

    @PostMapping("/add1")
    public Mono<Integer> add1(@RequestBody User user){
        return userService.add1(user);
    }
}

1. 开始测试

当我在intellij idea中执行如下请求时，果然抛出了异常，没有记录插入到数据库中。

POST http://localhost:8081/user/add1
Cache-Control: no-cache
Content-Type: application/json; charset=UTF-8

{
  "username": "1exception11",
  "password": "123456"
}

当我在intellij idea中执行如下请求时，如预期的那样，这条记录被插入到数据库中。

POST http://localhost:8081/user/add1
Cache-Control: no-cache
Content-Type: application/json; charset=UTF-8

{
  "username": "111",
  "password": "123456"
}

1. 以上代码是正确的步骤，也是我调试成功的最后一个版本。下面的代码是我一开始使用的，但是结果和我预想的不太一样。

在上面代码的基础上，增加一个类:

@Configuration
@EnableTransactionManagement
public class DatabaseConfiguration extends AbstractR2dbcConfiguration {
    @NotNull
    @Override
    public ConnectionFactory connectionFactory() {
        String url = "mysql://root:123456@127.0.0.1:3306/webflux-r2dbc";
        return new JasyncConnectionFactory(new MySQLConnectionFactory(URLParser.INSTANCE.parseOrDie(url, StandardCharsets.UTF_8)));
    }

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

}

并修改UserService.java -add方法add2()

@Service
public class UserService {

    private final static Logger LOGGER = LoggerFactory.getLogger(UserService.class);
    private final static String EXCEPTION = "exception";

    @Resource
    R2dbcEntityTemplate r2dbcEntityTemplate;
    @Resource
    UserRepository userRepository;


    @Transactional(rollbackFor = Exception.class)
    public Mono<Integer> add1(User queryUser) {

        return this.r2dbcEntityTemplate.insert(User.class)
                .using(queryUser)
                .doOnSuccess(user -> {
                    if (!user.getUsername().contains(EXCEPTION)) {
                        LOGGER.info("=====================add normal=================");
                    } else {
                        LOGGER.error("=====================add exception=================");
                        throw new RuntimeException("add1 exception test............");
                    }

                })
                .map(User::getId);
    }

    @Transactional(rollbackFor = Exception.class)
    public Mono<Integer> add2(User queryUser) {

        return userRepository.save(queryUser).flatMap((Function<User, Mono<Integer>>) user -> {
            if (user.getUsername().contains(EXCEPTION)) {
                LOGGER.error("=====================add2 exception=================");
                throw new RuntimeException("test exception...");
            }
            return Mono.just(user.getId());
        });
    }
}

并修改UserController.java -add方法add2()

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

    @Autowired
    private UserService userService;

    /**
     * transaction work
     * @param user
     * @return
     */

    @PostMapping("/add1")
    public Mono<Integer> add1(@RequestBody User user){
        return userService.add1(user);
    }


    /**
     * transaction not ork
     * @param user
     * @return
     */
    @PostMapping("/add2")
    public Mono<Integer> add2(@RequestBody User user){
        return userService.add2(user);
    }
}

测试:

当我在intellij idea中执行如下请求时，如预期的那样，这条记录被插入到数据库中。

POST http://localhost:8081/user/add1
Cache-Control: no-cache
Content-Type: application/json; charset=UTF-8

{
  "username": "222",
  "password": "123456"
}

当我在intellij idea中执行如下请求时，抛出异常，并将这条记录插入到数据库中。

POST http://localhost:8081/user/add1
Cache-Control: no-cache
Content-Type: application/json; charset=UTF-8

{
  "username": "222exception",
  "password": "123456"
}

为什么会这样，谁能解释一下，非常感谢。

Answer 1

我遇到了同样的问题，并且能够通过注册 R2dbcEntityOperations bean 并将此 bean 放入 @EnableR2dbcRepositories=entityOperationsRef = "yourBeanName") 来解决这个问题/p>