使用 Spring Data Reactive 与 Couchbase 构建响应式数据库操作

1. 概述

本文将带你掌握如何在 Spring Data 中配置并实现基于 Couchbase 的响应式数据库操作。

我们将重点使用 ReactiveCrudRepository 和 ReactiveSortingRepository 两个核心接口完成常见数据访问。同时，通过继承 AbstractReactiveCouchbaseConfiguration 来完成响应式连接的初始化配置。

整个过程踩坑不多，但有几个关键点需要注意，比如 N1QL 支持、视图（View）配置等，下文会一一说明。

2. Maven 依赖

首先，引入必要的依赖项：

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-couchbase-reactive</artifactId>
</dependency>

✅ 说明：

spring-boot-starter-data-couchbase-reactive：这是核心依赖，包含了 Spring Data Couchbase 的响应式支持，自动集成 Reactor 流。
reactor-core：提供 Mono 和 Flux 等响应式类型，是响应式编程的基石。

⚠️ 注意：如果你使用的是较老版本的 Spring Boot，可能需要手动指定 Couchbase SDK 版本，避免兼容性问题。

3. 配置响应式连接

要让 Spring 能以响应式方式连接 Couchbase，必须继承 AbstractReactiveCouchbaseConfiguration，而不是普通的 AbstractCouchbaseConfiguration。

3.1 配置属性类

先定义一个配置类来承载连接参数：

@Configuration
public class CouchbaseProperties {

    private List<String> bootstrapHosts;
    private String bucketName;
    private String bucketPassword;
    private int port;

    public CouchbaseProperties(
      @Value("${spring.couchbase.bootstrap-hosts}") List<String> bootstrapHosts, 
      @Value("${spring.couchbase.bucket.name}") String bucketName, 
      @Value("${spring.couchbase.bucket.password}") String bucketPassword, 
      @Value("${spring.couchbase.port}") int port) {
        this.bootstrapHosts = Collections.unmodifiableList(bootstrapHosts);
        this.bucketName = bucketName;
        this.bucketPassword = bucketPassword;
        this.port = port;
    }

    // getters
}

假设你的 application.yml 配置如下：

spring:
  couchbase:
    bootstrap-hosts:
      - localhost
    bucket:
      name: users
      password: secret123
    port: 8091

3.2 响应式配置类

接下来是关键的配置类：

@Configuration
@EnableReactiveCouchbaseRepositories(basePackages = "com.example.repo")
public abstract class ReactiveCouchbaseConfiguration extends AbstractReactiveCouchbaseConfiguration {

    private CouchbaseProperties couchbaseProperties;

    public ReactiveCouchbaseConfiguration(final CouchbaseProperties couchbaseProperties) {
        this.couchbaseProperties = couchbaseProperties;
    }

    @Override
    public String getConnectionString() {
        return String.join(",", couchbaseProperties.getBootstrapHosts());
    }

    @Override
    public String getUserName() {
        return couchbaseProperties.getBucketName();
    }

    @Override
    public String getPassword() {
        return couchbaseProperties.getBucketPassword();
    }

    @Override
    public String getBucketName() {
        return couchbaseProperties.getBucketName();
    }
}

✅ 关键点解析：

@EnableReactiveCouchbaseRepositories：启用响应式仓库支持，必须指定 basePackages 扫描路径。
继承 AbstractReactiveCouchbaseConfiguration：这是响应式的前提，底层使用异步 Couchbase Client。
方法重写：将配置属性注入到连接中，确保 Spring 能正确建立响应式连接。

❌ 踩坑提醒：
如果忘记继承 AbstractReactiveCouchbaseConfiguration，而是用了普通配置类，会导致仓库方法返回 Mono 但底层仍是阻塞调用，失去响应式意义。

4. 响应式仓库（Repository）使用

Spring Data Reactive 支持两种查询机制：

✅ View-based：基于 Couchbase 的 View（视图），适合简单查询，性能较好。
✅ N1QL-based：基于 SQL-like 的 N1QL 查询，功能强大，支持复杂条件和排序。

⚠️ 注意：默认情况下，CRUD 操作由 all 视图支持。自定义方法若使用 N1QL，需确保集群已启用 N1QL 服务（即创建了 Primary Index），否则会抛出 UnsupportedCouchbaseFeatureException。

4.1 实体类定义

先定义一个简单的 Person 实体：

@Document
public class Person {
    @Id private UUID id;
    private String firstName;

    // 构造函数、getters、setters 省略
}

@Document：标识这是一个 Couchbase 文档实体。
@Id：指定主键字段，对应文档的 ID。

4.2 基于 View 的仓库

适用于简单查询，需预先在 Couchbase 中创建 Design Document 和 View。

@Repository
@ViewIndexed(designDoc = ViewPersonRepository.DESIGN_DOCUMENT)
public interface ViewPersonRepository extends ReactiveCrudRepository<Person, UUID> {

    String DESIGN_DOCUMENT = "person";

    @View(designDocument = DESIGN_DOCUMENT)
    Flux<Person> findByFirstName(String firstName);
}

@ViewIndexed：声明该仓库使用的 Design Document 名称为 person。
@View：标注方法使用 View 查询，默认查找名为 byFirstName 的 View。
若 View 名不同，可用 @View(viewName = "customView") 指定。

✅ 测试示例：

@Test
public void shouldSavePerson_findById_thenDeleteIt() {
    final UUID id = UUID.randomUUID();
    final Person person = new Person(id, "John");
    personRepository.save(person).subscribe();

    final Mono<Person> byId = personRepository.findById(id);

    StepVerifier.create(byId)
        .expectNextMatches(result -> result.getId().equals(id))
        .expectComplete()
        .verify();

    personRepository.delete(person).subscribe();
}

使用 StepVerifier 验证响应式流行为，这是响应式测试的标准做法。

4.3 基于 N1QL 的仓库

更灵活，支持方法名解析生成 N1QL 查询。

@Repository
public interface N1QLPersonRepository extends ReactiveCrudRepository<Person, UUID> {
    Flux<Person> findAllByFirstName(String firstName);
}

方法名 findAllByFirstName 会被自动解析为 N1QL 查询：
SELECT * FROM bucket_name WHERE firstName = $1

✅ 测试代码：

@Test
public void shouldFindAll_byFirstName() {
    final String firstName = "John";
    final Person matchingPerson = new Person(UUID.randomUUID(), firstName);
    final Person nonMatchingPerson = new Person(UUID.randomUUID(), "NotJohn");

    personRepository.save(matchingPerson).subscribe();
    personRepository.save(nonMatchingPerson).subscribe();

    final Flux<Person> result = personRepository.findAllByFirstName(firstName);

    StepVerifier.create(result)
        .expectNext(matchingPerson)
        .verifyComplete();
}

4.4 支持排序的仓库

通过继承 ReactiveSortingRepository，可支持排序功能。

@Repository
public interface N1QLSortingPersonRepository 
    extends ReactiveCrudRepository<Person, UUID>, 
            ReactiveSortingRepository<Person, UUID> {
}

✅ 测试排序功能：

@Test
public void shouldFindAll_sortedByFirstName() {
    final Person firstPerson = new Person(UUID.randomUUID(), "John");
    final Person secondPerson = new Person(UUID.randomUUID(), "Mikki");

    personRepository.save(firstPerson).subscribe();
    personRepository.save(secondPerson).subscribe();

    final Flux<Person> sorted = personRepository.findAll(Sort.by(Sort.Direction.DESC, "firstName"));

    StepVerifier.create(sorted)
        .expectNextMatches(person -> person.getFirstName().equals("Mikki"))
        .expectNextMatches(person -> person.getFirstName().equals("John"))
        .verifyComplete();
}

排序由 N1QL 的 ORDER BY 实现，简单粗暴有效。

5. 总结

本文系统介绍了如何在 Spring Boot 项目中使用 Spring Data Reactive 与 Couchbase 构建响应式数据访问层。

✅ 核心要点回顾：

使用 spring-boot-starter-data-couchbase-reactive 启用响应式支持。
配置类必须继承 AbstractReactiveCouchbaseConfiguration。
仓库接口可选择 View 或 N1QL 模式，后者更灵活但需启用索引。
自定义查询方法命名遵循 Spring Data 规范，自动映射为 N1QL。
测试使用 StepVerifier 验证 Mono/Flux 行为。

⚠️ 生产建议：

N1QL 查询前确保创建了 Primary Index：CREATE PRIMARY INDEX ON bucket_name
高频查询字段建议创建 Secondary Index 提升性能。
响应式流中错误处理要完善，避免流中断。

所有示例代码已托管至 GitHub：
https://github.com/baeldung/spring-reactive-data-couchbase

Persistence

REST

Security