为什么在自定义 AbstractR2dbcConfiguration

如何解决为什么在自定义 AbstractR2dbcConfiguration

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

您可以在 master 分支上看到正确的代码，在 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>

application.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

演示应用

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

实体类

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

用户存储库：

public interface UserRepository extends ReactiveCrudRepository<User,Integer> {

}

用户服务

@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);
    }
}

开始测试

当我在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"
}

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

在以上代码的基础上，添加一个类：

@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"
}

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

为什么在自定义 AbstractR2dbcConfiguration

如何解决为什么在自定义 AbstractR2dbcConfiguration

相关推荐