如何在PostgreSQL中正确使用Spring R2DBC DatabaseClient执行自定义删除查询

核心问题：SQL未实际执行 #

你当前的deleteWithCustomSql方法存在致命问题：仅调用databaseClient.execute(sql)创建了执行对象，但Reactive操作需要订阅才会触发执行，而你直接返回Mono.empty()，导致删除语句根本没发送到数据库。

正确实现方式 #

1. 修复执行逻辑，触发SQL执行 #

修改CustomSqlService的方法，通过fetch().rowsUpdated()触发SQL执行，并转换为Mono<Void>表示操作完成：

import org.springframework.data.r2dbc.core.DatabaseClient;
import reactor.core.publisher.Mono;

@Service
@Transactional
public class CustomSqlService {
    private final DatabaseClient databaseClient;

    // 构造函数注入DatabaseClient
    public CustomSqlService(DatabaseClient databaseClient) {
        this.databaseClient = databaseClient;
    }

    public Mono<Void> deleteWithCustomSql(String sql) {
        return databaseClient.execute(sql)
                .fetch() // 触发SQL执行，获取结果
                .rowsUpdated() // 返回受影响行数（Mono<Long>）
                .then(); // 转换为Mono<Void>，标识操作完成
    }
}

2. 规避SQL注入：改用参数绑定 #

直接拼接SQL字符串存在严重的SQL注入风险，必须使用参数绑定方式：

// 在CustomSqlService中新增带参数的安全删除方法
public Mono<Void> deleteUserById(Long userId) {
    String sql = "DELETE FROM user_table WHERE user_id = :userId";
    return databaseClient.execute(sql)
            .bind("userId", userId) // 绑定参数，避免注入
            .fetch()
            .rowsUpdated()
            .then();
}

// 控制器对应修改
@DeleteMapping(path = "/delete_user/{userId}")
public Mono<Void> deleteUser(@PathVariable("userId") Long userId)  {
   return customSqlService.deleteUserById(userId);
}

3. Reactive事务注意事项 #

• 确保Spring环境已正确配置Reactive事务管理器（如R2dbcTransactionManager），否则@Transactional注解不会生效。
• 事务会作用于整个Reactive流，确保删除操作在事务上下文内执行。

测试代码优化 #

测试时需确保删除操作完全完成后再执行查询验证，可通过expectBody()等待删除流结束：

@Test
void shouldDeleteDataFromTable() {
    User user = User.builder()
            .userId(1L)
            .sessionId(2L)
            .timestamp(10L)
            .build();

    // 插入用户
    webTestClient
            .post()
            .uri("/postgres/save_user")
            .contentType(MediaType.APPLICATION_JSON)
            .accept(MediaType.APPLICATION_JSON)
            .body(Mono.just(user), User.class)
            .exchange()
            .expectStatus().isOk();

    // 执行删除并等待操作完成
    webTestClient
            .delete()
            .uri("/postgres/delete_user/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody(Void.class);

    // 查询验证数据已删除
    userRepository.findAll()
            .as(StepVerifier::create)
            .expectNextCount(0)
            .verifyComplete();
}

错误排查补充 #

若仍出现MonoOnResumeError，可通过doOnError捕获并打印具体异常信息，定位问题根源：

public Mono<Void> deleteUserById(Long userId) {
    String sql = "DELETE FROM user_table WHERE user_id = :userId";
    return databaseClient.execute(sql)
            .bind("userId", userId)
            .fetch()
            .rowsUpdated()
            .doOnError(e -> System.err.println("删除失败：" + e.getMessage()))
            .then();
}

内容的提问来源于stack exchange，提问作者Baal