Spring Data R2DBC中@Query注解与Flux参数的使用限制

本文深入探讨了Spring Data R2DBC中，当自定义仓库方法结合`@Query`注解并以`Flux`作为参数时遇到的问题。核心问题在于`@Query`注解无法自动订阅并处理传入的`Flux`参数，导致参数绑定失败并抛出`IllegalArgumentException: Value must not be null`。文章提供了详细的代码示例，分析了错误产生的根源，并指出解决方案是利用Spring Data R2DBC的派生查询机制，避免在处理`Flux`参数时使用`@Query`注解。

引言：Spring Data R2DBC与自定义查询

Spring Data R2DBC为响应式关系型数据库访问提供了强大的支持，通过抽象层简化了数据操作。开发者可以定义继承自ReactiveCrudRepository的接口来自动获得基本的CRUD操作，同时也可以通过方法命名约定（派生查询）或使用@Query注解来定义自定义查询。@Query注解允许开发者直接编写SQL语句，以满足更复杂的查询需求。然而，在使用@Query注解结合响应式流类型（如Flux）作为方法参数时，可能会遇到一些意料之外的行为。

问题描述：@Query注解与Flux参数的冲突

当尝试在Spring Data R2DBC的自定义仓库方法中使用@Query注解，并且该方法接受一个Flux类型的参数时，系统会抛出IllegalArgumentException: Value must not be null异常。以下是一个典型的示例，展示了这种配置及其导致的错误：

实体定义：

@Getter
@Setter
@ToString
@Data
@NoArgsConstructor
@AllArgsConstructor
@Table(value = "Person", schema = "mySchema")
public class Person {
    @Id
    @Column("id")
    Long id;

    @Column("name")
    String name;
}

仓库接口：

public interface PersonRepository extends ReactiveCrudRepository {

    // 期望通过Flux参数查询
    @Query("SELECT id, name  FROM Person WHERE id = ?")
    Flux myMethod(Flux person); // 这里的person对象中的id字段被期望用于绑定
}

应用启动类中的调用示例：

@SpringBootApplication
@EnableR2dbcAuditing
@EnableConfigurationProperties({ApplicationProperties.class})
public class MyApplication {

    private static final Logger logger = LoggerFactory.getLogger(MyApplication.class);

    @Bean
    public CommandLineRunner consume(PersonRepository personRepository) {
        LongSupplier randomLong = () -> RandomUtils.nextLong(10L, 20L);
        Flux personIds = Flux.fromStream(LongStream.generate(randomLong).boxed());
        Flux persons = personIds.map( p -> {
            Person person = new Person();
            person.setId(p);
            return person;
        });

        // 调用自定义方法
        personRepository.myMethod(persons.take(3)).subscribe(id -> logger.info("Processed Person Id: " + id));

        return null;
    }   
}

执行上述代码时，会遇到以下异常：

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalArgumentException: Value must not be null
Caused by: java.lang.IllegalArgumentException: **Value must not be null**
    at org.springframework.util.Assert.notNull(Assert.java:201)
    at org.springframework.r2dbc.core.Parameter.from(Parameter.java:54)
    // ... (省略部分堆栈信息)

这个异常明确指出在参数绑定过程中，尝试绑定的值是null，这表明Spring Data R2DBC的@Query机制未能正确地从传入的Flux参数中提取出用于SQL绑定的实际值。

根源分析：@Query注解的内部机制

问题根源在于@Query注解在处理方法参数时，通常期望接收一个单一的、可直接用于绑定的值，而不是一个响应式流（Publisher，如Flux或Mono）。当@Query遇到一个Flux类型的参数时，它并不会自动订阅这个Flux来获取其发出的元素，然后将每个元素逐一绑定到SQL查询中。相反，它会将整个Flux对象本身视为一个待绑定的参数。由于Flux对象本身不是一个可直接映射到数据库列的简单值（如Long、String等），在尝试将其转换为Parameter时，内部处理逻辑无法从中提取出有效的绑定值，最终导致null值被传递，触发IllegalArgumentException。

这与Spring Data R2DBC内置的findAllById(Publisher idStream)或findById(Publisher id)等方法形成对比。这些内置方法是专门设计来处理Publisher参数的，它们会内部订阅传入的流，并针对流中的每个元素执行查询或批量查询。而@Query注解则不具备这种自动订阅和扁平化（flatMap）Publisher参数的能力。

Quinvio AI

AI辅助下快速创建视频，虚拟代言人

下载

解决方案：利用Spring Data的派生查询

解决此问题的最直接和推荐方法是移除@Query注解，转而依赖Spring Data R2DBC的派生查询机制。Spring Data能够根据方法名自动推断出查询意图，并且它对处理Flux类型的参数有着良好的支持，尤其是在进行批量查询时。

例如，如果你的目标是根据一系列ID查询多个Person对象，你可以这样定义你的仓库方法：

修改后的仓库接口：

public interface PersonRepository extends ReactiveCrudRepository {

    // 移除@Query注解，Spring Data将根据方法名推断查询
    // 自动处理Flux参数，执行类似findAllById的行为
    Flux findAllById(Flux ids); // 注意：这里的参数类型应是ID的类型，而非Person对象

    // 如果是根据其他字段批量查询，例如name
    Flux findAllByName(Flux names);
}

应用启动类中的调用示例（以ID为例）：

@SpringBootApplication
@EnableR2dbcAuditing
@EnableConfigurationProperties({ApplicationProperties.class})
public class MyApplication {

    private static final Logger logger = LoggerFactory.getLogger(MyApplication.class);

    @Bean
    public CommandLineRunner consume(PersonRepository personRepository) {
        LongSupplier randomLong = () -> RandomUtils.nextLong(10L, 20L);
        Flux personIds = Flux.fromStream(LongStream.generate(randomLong).boxed());

        // 调用派生查询方法
        personRepository.findAllById(personIds.take(3)) // 传入Flux
            .subscribe(person -> logger.info("Processed Person: " + person));

        return null;
    }   
}

通过这种方式，Spring Data R2DBC会识别findAllById方法，并正确地订阅传入的Flux，然后为流中的每个ID执行查询，或者在可能的情况下进行批量查询优化。

替代方案与注意事项

尽管派生查询是处理Flux参数的首选方式，但在某些极端情况下，如果你确实需要一个高度定制的SQL查询，并且需要根据多个值进行过滤（例如使用SQL的IN子句），你可以考虑以下策略：

1. 使用Collection参数与IN子句： 如果你的自定义查询需要根据一组ID进行过滤，通常的做法是将Flux转换为Mono>，然后将List作为参数传递给带有IN子句的@Query方法。

   public interface PersonRepository extends ReactiveCrudRepository {
       @Query("SELECT id, name FROM Person WHERE id IN (:ids)")
       Flux findByListOfIds(@Param("ids") Collection ids);
   }
   
   // 调用时：
   personIds.collectList() // 将Flux转换为Mono>
            .flatMapMany(idList -> personRepository.findByListOfIds(idList))
            .subscribe(person -> logger.info("Processed Person: " + person));

   这种方法将Flux的扁平化处理逻辑移到了调用方，确保@Query注解接收到的是一个可以直接绑定的Collection类型。

2. 避免在@Query中直接绑定复杂对象： 在原始问题中，尝试将Flux绑定到?占位符，期望其内部的id字段被提取。@Query注解不支持这种深层对象属性的自动提取和绑定，尤其当参数本身是一个响应式流时。如果需要根据Person对象的某个属性查询，应明确提取该属性并作为参数传递。

总结

Spring Data R2DBC的@Query注解是一个强大的工具，用于执行自定义SQL查询。然而，它在处理Flux等响应式流类型作为方法参数时存在限制，即不会自动订阅并扁平化这些流以获取绑定值。当需要根据一系列值进行查询时，推荐使用Spring Data的派生查询机制，例如findAllById(Flux ids)，它能够正确地处理Flux参数。如果必须使用@Query进行复杂查询并涉及多个参数，应考虑将Flux转换为Collection（如List），并通过IN子句进行绑定，将响应式流的处理逻辑前置到调用方。理解这些限制和最佳实践，有助于更高效、更稳定地构建响应式数据访问层。