全栈攻城狮

Appearance

R2DBC 响应式数据库 ​

概述 ​

R2DBC（Reactive Relational Database Connectivity）是 Spring 生态中用于构建响应式数据访问层的规范，填补了 JDBC 在响应式编程中的空白。JDBC 本质上是阻塞 I/O 模型，而 R2DBC 基于 Reactive Streams，允许数据库操作在非阻塞线程模型上执行，完美契合 WebFlux 响应式架构。

WebFlux (Netty) → R2DBC SPI → R2DBC Driver → Database

核心依赖 ​

yaml
dependencies:
  - spring-boot-starter-webflux
  - spring-boot-starter-data-r2dbc
  - io.r2dbc:r2dbc-postgresql
  - io.r2dbc:r2dbc-pool

1. 数据源与连接池配置 ​

基础配置 ​

yaml
spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/bullet_screen
    username: postgres
    password: postgres
    pool:
      initial-size: 10
      max-size: 30
      max-idle-time: 30m
      validation-query: SELECT 1

编程式连接池调优 ​

java
@Configuration
public class R2dbcPoolConfig {

    @Bean
    public ConnectionPool connectionPool() {
        PostgresqlConnectionFactory factory = new PostgresqlConnectionFactory(
            PostgresqlConnectionConfiguration.builder()
                .host("localhost").port(5432).database("bullet_screen")
                .username("postgres").password("postgres").build());

        ConnectionPoolConfiguration poolConfig = ConnectionPoolConfiguration.builder(factory)
            .name("r2dbc-pool")
            .initialSize(10)                          // 初始连接数
            .maxSize(30)                              // 最大连接数
            .maxIdleTime(Duration.ofMinutes(30))      // 最大空闲时间
            .maxLifeTime(Duration.ofHours(2))         // 连接最大存活
            .maxAcquireTime(Duration.ofSeconds(5))    // 获取连接超时
            .maxCreateConnectionTime(Duration.ofSeconds(5))
            .validationQuery("SELECT 1")
            .build();

        return new ConnectionPool(poolConfig);
    }
}

连接池调优参数 ​

参数建议值说明
initialSizeCPU 核心数 × 2避免冷启动
maxSize50 ~ 100根据数据库上限调整
maxIdleTime10 ~ 30 min释放空闲连接
maxAcquireTime≤ 5 s避免请求堆积
validationQuerySELECT 1心跳检测

2. 实体映射与 Converter ​

基础实体 ​

java
@Table("bullet_comments")
public class BulletComment {

    @Id
    private Long id;
    private String userId;
    private String nickname;
    private String content;
    private Integer color;          // 弹幕颜色 ARGB
    private Integer fontSize;       // 字体大小
    private Double position;        // 弹幕在视频中的位置（秒）
    private String videoId;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;

    public BulletComment() {}
    // getters / setters（R2DBC 反射需要，省略以节省篇幅）
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getNickname() { return nickname; }
    public void setNickname(String nickname) { this.nickname = nickname; }
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
    public Integer getColor() { return color; }
    public void setColor(Integer color) { this.color = color; }
    public Integer getFontSize() { return fontSize; }
    public void setFontSize(Integer fontSize) { this.fontSize = fontSize; }
    public Double getPosition() { return position; }
    public void setPosition(Double position) { this.position = position; }
    public String getVideoId() { return videoId; }
    public void setVideoId(String videoId) { this.videoId = videoId; }
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    public LocalDateTime getUpdatedAt() { return updatedAt; }
    public void setUpdatedAt(LocalDateTime updatedAt) { this.updatedAt = updatedAt; }
}

自定义 Converter ​

java
@ReadingConverter
public class StatusReadConverter implements Converter<String, CommentStatus> {
    @Override
    public CommentStatus convert(String source) {
        return CommentStatus.valueOf(source.toUpperCase());
    }
}

@WritingConverter
public class StatusWriteConverter implements Converter<CommentStatus, String> {
    @Override
    public String convert(CommentStatus source) {
        return source.name().toLowerCase();
    }
}

注册 Converter ​

java
@Configuration
@EnableR2dbcRepositories
public class R2dbcConverterConfig {

    @Bean
    public R2dbcCustomConversions r2dbcCustomConversions(R2dbcDialect dialect) {
        return R2dbcCustomConversions.of(dialect, List.of(
            new StatusReadConverter(),
            new StatusWriteConverter()
        ));
    }
}

3. DatabaseClient 编程式 API ​

DatabaseClient 是 Spring Data R2DBC 的核心编程式 API，支持完全手动的 SQL 执行。

配置 ​

java
@Configuration
public class DatabaseClientConfig {

    @Bean
    public DatabaseClient databaseClient(ConnectionPool pool) {
        return DatabaseClient.builder()
            .connectionFactory(pool)
            .namedParameters(true)      // 启用 :name 语法
            .build();
    }
}

参数绑定方式 ​

java
// 按名称绑定
client.sql("SELECT * FROM t WHERE id = :id").bind("id", 1L).fetch().one();

// 按索引绑定（$1, $2）
client.sql("SELECT * FROM t WHERE id = $1").bind(0, 1L).fetch().one();

// 绑定 Bean 属性
client.sql("INSERT INTO t (id, name) VALUES (:id, :name)")
    .bindProperties(myBean).fetch().one();

CRUD 操作示例 ​

java
@Repository
public class BulletCommentDao {

    private final DatabaseClient client;

    public BulletCommentDao(DatabaseClient client) {
        this.client = client;
    }

    // Create
    public Mono<BulletComment> insert(BulletComment c) {
        return client.sql("""
                INSERT INTO bullet_comments
                    (user_id, nickname, content, color, font_size, position, video_id, created_at, updated_at)
                VALUES (:userId, :nickname, :content, :color, :fontSize, :position, :videoId, :createdAt, :updatedAt)
                RETURNING *
                """)
            .bind("userId", c.getUserId()).bind("nickname", c.getNickname())
            .bind("content", c.getContent()).bind("color", c.getColor())
            .bind("fontSize", c.getFontSize()).bind("position", c.getPosition())
            .bind("videoId", c.getVideoId())
            .bind("createdAt", LocalDateTime.now()).bind("updatedAt", LocalDateTime.now())
            .fetch().one().map(this::toEntity);
    }

    // Read
    public Flux<BulletComment> findByVideoId(String videoId) {
        return client.sql("""
                SELECT * FROM bullet_comments
                WHERE video_id = :videoId ORDER BY position ASC
                """)
            .bind("videoId", videoId)
            .fetch().all().map(this::toEntity);
    }

    public Mono<BulletComment> findById(Long id) {
        return client.sql("SELECT * FROM bullet_comments WHERE id = :id")
            .bind("id", id).fetch().one().map(this::toEntity);
    }

    // Update
    public Mono<Integer> updateContent(Long id, String content) {
        return client.sql("UPDATE bullet_comments SET content = :content, updated_at = NOW() WHERE id = :id")
            .bind("id", id).bind("content", content)
            .fetch().rowsUpdated();
    }

    // Delete
    public Mono<Integer> deleteById(Long id) {
        return client.sql("DELETE FROM bullet_comments WHERE id = :id")
            .bind("id", id).fetch().rowsUpdated();
    }

    // 聚合
    public Mono<Long> countByVideoId(String videoId) {
        return client.sql("SELECT COUNT(*) AS cnt FROM bullet_comments WHERE video_id = :videoId")
            .bind("videoId", videoId).fetch().one()
            .map(row -> ((Number) row.get("cnt")).longValue());
    }

    // 批量插入
    public Flux<BulletComment> batchInsert(List<BulletComment> comments) {
        return Flux.fromIterable(comments).flatMap(this::insert);
    }

    private BulletComment toEntity(Map<String, Object> row) {
        BulletComment c = new BulletComment();
        c.setId(((Number) row.get("id")).longValue());
        c.setUserId((String) row.get("user_id"));
        c.setNickname((String) row.get("nickname"));
        c.setContent((String) row.get("content"));
        c.setColor((Integer) row.get("color"));
        c.setFontSize((Integer) row.get("font_size"));
        c.setPosition((Double) row.get("position"));
        c.setVideoId((String) row.get("video_id"));
        c.setCreatedAt((LocalDateTime) row.get("created_at"));
        c.setUpdatedAt((LocalDateTime) row.get("updated_at"));
        return c;
    }
}

4. R2dbcEntityTemplate CRUD ​

基于实体的更高层模板 API，无需手写 SQL。

配置 ​

java
@Bean
public R2dbcEntityTemplate r2dbcEntityTemplate(ConnectionPool pool) {
    return new R2dbcEntityTemplate(pool);
}

CRUD 操作 ​

java
import static org.springframework.data.relational.core.query.Criteria.where;
import static org.springframework.data.relational.core.query.Query.query;
import static org.springframework.data.relational.core.query.Update.update;

@Repository
public class BulletCommentTemplateDao {

    private final R2dbcEntityTemplate template;

    public BulletCommentTemplateDao(R2dbcEntityTemplate template) {
        this.template = template;
    }

    public Mono<BulletComment> save(BulletComment c) {
        return template.insert(c);
    }

    public Flux<BulletComment> findByVideoId(String videoId) {
        return template.select(BulletComment.class)
            .matching(query(where("video_id").is(videoId)))
            .all();
    }

    public Mono<BulletComment> findById(Long id) {
        return template.selectOne(query(where("id").is(id)), BulletComment.class);
    }

    public Mono<Integer> updateContent(Long id, String content) {
        return template.update(BulletComment.class)
            .matching(query(where("id").is(id)))
            .apply(update("content", content));
    }

    public Mono<Integer> deleteById(Long id) {
        return template.delete(BulletComment.class)
            .matching(query(where("id").is(id)))
            .all();
    }

    public Mono<Boolean> existsById(Long id) {
        return template.exists(query(where("id").is(id)), BulletComment.class);
    }

    public Mono<Long> countByVideoId(String videoId) {
        return template.select(BulletComment.class)
            .matching(query(where("video_id").is(videoId)))
            .count();
    }
}

Criteria 进阶 ​

java
import org.springframework.data.domain.Sort;

// 多条件组合
Criteria criteria = where("video_id").is(videoId)
    .and("position").between(0.0, 60.0)
    .and("color").not(0xFFFF0000);

// 排序分页
Flux<BulletComment> result = template.select(BulletComment.class)
    .matching(query(criteria)
        .sort(Sort.by(Sort.Direction.ASC, "position"))
        .offset(0).limit(20))
    .all();

5. @Query 声明式查询 ​

Spring Data R2DBC 的 Repository 通过 @Query 注解将 SQL 声明在接口方法上。

Repository 定义 ​

java
public interface BulletCommentRepository extends R2dbcRepository<BulletComment, Long> {

    // 声明式 SQL 查询
    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId ORDER BY position ASC")
    Flux<BulletComment> findByVideoId(@Param("videoId") String videoId);

    // 分页
    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId ORDER BY created_at DESC LIMIT :limit OFFSET :offset")
    Flux<BulletComment> findByVideoIdPaged(@Param("videoId") String videoId,
                                           @Param("limit") int limit,
                                           @Param("offset") int offset);

    // 统计
    @Query("SELECT COUNT(*) FROM bullet_comments WHERE video_id = :videoId")
    Mono<Long> countByVideoId(@Param("videoId") String videoId);

    // 更新
    @Query("UPDATE bullet_comments SET content = :content, updated_at = NOW() WHERE id = :id")
    Mono<Integer> updateContent(@Param("id") Long id, @Param("content") String content);

    // 删除
    @Query("DELETE FROM bullet_comments WHERE id = :id")
    Mono<Integer> deleteCommentById(@Param("id") Long id);

    // 时间段查询
    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId AND created_at BETWEEN :start AND :end ORDER BY created_at DESC")
    Flux<BulletComment> findBetween(@Param("videoId") String videoId,
                                    @Param("start") LocalDateTime start,
                                    @Param("end") LocalDateTime end);

    // 全文检索（PostgreSQL tsvector）
    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId AND to_tsvector('simple', content) @@ plainto_tsquery('simple', :keyword) ORDER BY position ASC")
    Flux<BulletComment> searchByKeyword(@Param("videoId") String videoId, @Param("keyword") String keyword);
}

方法命名查询 ​

java
public interface BulletCommentRepository extends R2dbcRepository<BulletComment, Long> {

    Flux<BulletComment> findByVideoIdOrderByPositionAsc(String videoId);
    Flux<BulletComment> findByUserId(String userId);
    Flux<BulletComment> findByVideoIdAndColor(String videoId, Integer color);
    Mono<Long> countByVideoId(String videoId);
    Mono<Boolean> existsByVideoIdAndUserId(String videoId, String userId);
    Mono<Integer> deleteByVideoId(String videoId);
}

自定义 Repository ​

java
public interface CustomBulletCommentRepository {
    Flux<BulletComment> findHotComments(String videoId, int limit);
}

public class CustomBulletCommentRepositoryImpl implements CustomBulletCommentRepository {

    private final DatabaseClient client;

    public CustomBulletCommentRepositoryImpl(DatabaseClient client) {
        this.client = client;
    }

    @Override
    public Flux<BulletComment> findHotComments(String videoId, int limit) {
        return client.sql("SELECT * FROM bullet_comments WHERE video_id = :videoId ORDER BY RANDOM() LIMIT :limit")
            .bind("videoId", videoId).bind("limit", limit)
            .fetch().all()
            .map(row -> { /* 映射逻辑 */ return new BulletComment(); });
    }
}

6. 事务 ReactiveTransactionManager ​

配置 ​

java
@Configuration
@EnableTransactionManagement
public class R2dbcTransactionConfig {

    @Bean
    public ReactiveTransactionManager transactionManager(ConnectionPool pool) {
        return new R2dbcTransactionManager(pool);
    }
}

声明式事务 ​

java
@Service
public class BulletCommentService {

    private final BulletCommentRepository repository;
    private final UserScoreRepository scoreRepository;

    public BulletCommentService(BulletCommentRepository repository,
                                UserScoreRepository scoreRepository) {
        this.repository = repository;
        this.scoreRepository = scoreRepository;
    }

    // 单表事务
    @Transactional
    public Mono<BulletComment> createComment(BulletComment comment) {
        return repository.save(comment);
    }

    // 跨表事务：发弹幕 + 加积分
    @Transactional
    public Mono<BulletComment> createCommentWithScore(BulletComment comment) {
        return repository.save(comment)
            .flatMap(saved -> scoreRepository.increaseScore(saved.getUserId(), 1)
                .thenReturn(saved));
    }

    // 超时事务
    @Transactional(timeout = 5)
    public Flux<BulletComment> batchCreate(List<BulletComment> comments) {
        return repository.saveAll(comments);
    }
}

编程式事务 ​

java
@Service
public class TransactionalOperatorService {

    private final TransactionalOperator operator;
    private final BulletCommentRepository repository;

    public TransactionalOperatorService(TransactionalOperator operator,
                                        BulletCommentRepository repository) {
        this.operator = operator;
        this.repository = repository;
    }

    public Flux<BulletComment> batchInsertWithTx(List<BulletComment> comments) {
        return operator.execute(status -> repository.saveAll(comments));
    }

    public Mono<BulletComment> createWithRollback(BulletComment comment) {
        return operator.execute(status ->
            repository.save(comment).flatMap(saved -> {
                if (saved.getContent().contains("spam")) {
                    status.setRollbackOnly();
                    return Mono.error(new RuntimeException("Spam rejected"));
                }
                return Mono.just(saved);
            }));
    }
}

事务注意事项 ​

场景建议
只读操作@Transactional(readOnly = true) 优化连接
大事务拆分为小事务，避免长连接
跨库事务R2DBC 不支持分布式事务，用 Saga 模式
阻塞操作避免在事务中混入 restTemplate 等阻塞调用

7. 实战：WebFlux + R2DBC + PostgreSQL 实时弹幕系统 ​

建表 SQL ​

sql
CREATE TABLE IF NOT EXISTS bullet_comments (
    id          BIGSERIAL       PRIMARY KEY,
    user_id     VARCHAR(64)     NOT NULL,
    nickname    VARCHAR(64)     NOT NULL DEFAULT '匿名用户',
    content     TEXT            NOT NULL,
    color       INTEGER         NOT NULL DEFAULT 16777215,  -- 0xFFFFFF
    font_size   INTEGER         NOT NULL DEFAULT 28,
    position    DOUBLE PRECISION NOT NULL DEFAULT 0,        -- 弹幕在视频中的位置（秒）
    video_id    VARCHAR(128)    NOT NULL,
    created_at  TIMESTAMP       NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMP       NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_bullet_video_position ON bullet_comments (video_id, position);
CREATE INDEX idx_bullet_user ON bullet_comments (user_id);
CREATE INDEX idx_bullet_created ON bullet_comments (created_at DESC);

实体 ​

参见第 2 节的 BulletComment 实体类。

Repository ​

java
@Repository
public interface BulletCommentRepository extends R2dbcRepository<BulletComment, Long> {

    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId ORDER BY position ASC")
    Flux<BulletComment> findByVideoId(@Param("videoId") String videoId);

    @Query("SELECT * FROM bullet_comments WHERE video_id = :videoId AND position BETWEEN :startPos AND :endPos ORDER BY position ASC")
    Flux<BulletComment> findInRange(@Param("videoId") String videoId,
                                    @Param("startPos") Double startPos,
                                    @Param("endPos") Double endPos);

    Mono<Long> countByVideoId(String videoId);
    Mono<Integer> deleteByVideoId(String videoId);
}

DTO ​

java
public class BulletCommentReq {
    private String userId; private String nickname; private String content;
    private Integer color; private Integer fontSize; private Double position;
    private String videoId;
    // getters & setters
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    public String getNickname() { return nickname; }
    public void setNickname(String nickname) { this.nickname = nickname; }
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
    public Integer getColor() { return color; }
    public void setColor(Integer color) { this.color = color; }
    public Integer getFontSize() { return fontSize; }
    public void setFontSize(Integer fontSize) { this.fontSize = fontSize; }
    public Double getPosition() { return position; }
    public void setPosition(Double position) { this.position = position; }
    public String getVideoId() { return videoId; }
    public void setVideoId(String videoId) { this.videoId = videoId; }
}

public class BulletCommentResp {
    private Long id; private String userId; private String nickname;
    private String content; private Integer color; private Integer fontSize;
    private Double position; private String videoId; private LocalDateTime createdAt;

    public BulletCommentResp(BulletComment c) {
        this.id = c.getId(); this.userId = c.getUserId(); this.nickname = c.getNickname();
        this.content = c.getContent(); this.color = c.getColor(); this.fontSize = c.getFontSize();
        this.position = c.getPosition(); this.videoId = c.getVideoId();
        this.createdAt = c.getCreatedAt();
    }
    // getters
    public Long getId() { return id; }
    public String getUserId() { return userId; }
    public String getNickname() { return nickname; }
    public String getContent() { return content; }
    public Integer getColor() { return color; }
    public Integer getFontSize() { return fontSize; }
    public Double getPosition() { return position; }
    public String getVideoId() { return videoId; }
    public LocalDateTime getCreatedAt() { return createdAt; }
}

Service ​

java
@Service
@Transactional
public class BulletCommentService {

    private final BulletCommentRepository repository;

    public BulletCommentService(BulletCommentRepository repository) {
        this.repository = repository;
    }

    public Mono<BulletCommentResp> sendComment(BulletCommentReq req) {
        BulletComment c = new BulletComment();
        c.setUserId(req.getUserId());
        c.setNickname(req.getNickname() != null ? req.getNickname() : "匿名用户");
        c.setContent(req.getContent());
        c.setColor(req.getColor() != null ? req.getColor() : 0xFFFFFF);
        c.setFontSize(req.getFontSize() != null ? req.getFontSize() : 28);
        c.setPosition(req.getPosition() != null ? req.getPosition() : 0);
        c.setVideoId(req.getVideoId());
        return repository.save(c).map(BulletCommentResp::new);
    }

    @Transactional(readOnly = true)
    public Flux<BulletCommentResp> getComments(String videoId) {
        return repository.findByVideoId(videoId).map(BulletCommentResp::new);
    }

    @Transactional(readOnly = true)
    public Flux<BulletCommentResp> getCommentsInRange(String videoId, Double startPos, Double endPos) {
        return repository.findInRange(videoId, startPos, endPos).map(BulletCommentResp::new);
    }

    @Transactional(readOnly = true)
    public Mono<Long> getCommentCount(String videoId) {
        return repository.countByVideoId(videoId);
    }

    @Transactional(readOnly = true)
    public Mono<BulletCommentResp> getCommentById(Long id) {
        return repository.findById(id)
            .map(BulletCommentResp::new)
            .switchIfEmpty(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND, "评论不存在")));
    }

    public Mono<Void> deleteComment(Long id) {
        return repository.deleteById(id).then();
    }
}

Controller ​

java
@RestController
@RequestMapping("/api/bullet-comments")
public class BulletCommentController {

    private final BulletCommentService service;

    public BulletCommentController(BulletCommentService service) {
        this.service = service;
    }

    @PostMapping
    public Mono<BulletCommentResp> send(@RequestBody BulletCommentReq req) {
        return service.sendComment(req);
    }

    @GetMapping("/{videoId}")
    public Flux<BulletCommentResp> getComments(@PathVariable String videoId) {
        return service.getComments(videoId);
    }

    @GetMapping("/{videoId}/range")
    public Flux<BulletCommentResp> getInRange(@PathVariable String videoId,
                                              @RequestParam Double startPos,
                                              @RequestParam Double endPos) {
        return service.getCommentsInRange(videoId, startPos, endPos);
    }

    @GetMapping("/{videoId}/count")
    public Mono<Long> getCount(@PathVariable String videoId) {
        return service.getCommentCount(videoId);
    }

    @GetMapping("/detail/{id}")
    public Mono<BulletCommentResp> getById(@PathVariable Long id) {
        return service.getCommentById(id);
    }

    @DeleteMapping("/{id}")
    public Mono<Void> delete(@PathVariable Long id) {
        return service.deleteComment(id);
    }

    // SSE 实时推送
    @GetMapping(value = "/stream/{videoId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<BulletCommentResp>> stream(@PathVariable String videoId) {
        return Flux.interval(Duration.ofSeconds(2))
            .flatMap(tick -> service.getComments(videoId)
                .map(c -> ServerSentEvent.<BulletCommentResp>builder(c)
                    .event("bullet").id(String.valueOf(c.getId())).build()));
    }
}

完整配置 ​

yaml
server:
  port: 8080
spring:
  r2dbc:
    url: r2dbc:pool:postgresql://localhost:5432/bullet_screen
    username: postgres
    password: postgres
    pool:
      initial-size: 10
      max-size: 30
      max-idle-time: 30m
      validation-query: SELECT 1
logging:
  level:
    org.springframework.r2dbc: DEBUG

8. R2DBC vs JPA vs MyBatis 对比 ​

维度R2DBCJPA (Hibernate)MyBatis
I/O 模型响应式（非阻塞）阻塞阻塞
线程模型事件循环（少量线程）线程池（每个请求一个线程）线程池
适用场景高并发实时系统、流处理企业级 CRUD、复杂关联复杂 SQL、遗留数据库
学习曲线中等（需理解 Reactor）较高（JPA 规范复杂）低（只要 SQL）
SQL 控制力完全控制弱（自动生成）完全控制
自动建表不支持支持 (ddl-auto)不支持
一级缓存无（无状态设计）有 (PersistenceContext)无
懒加载不支持支持支持
N+1 问题不存在常见（需 fetch join）需手动处理
多表关联手动处理（不推荐关联）@OneToMany / @ManyToOne<association>
分页手动 LIMIT/OFFSETPageableRowBounds / 插件
性能吞吐量高（非阻塞）中等高（SQL 优化空间大）
社区生态较新（发展中）成熟稳定成熟稳定

选型建议 ​

  • R2DBC：全栈响应式架构、高吞吐实时系统（弹幕、消息推送、IoT）
  • JPA：业务模型复杂、关联关系多、团队熟悉 ORM
  • MyBatis：SQL 复杂需精细优化、遗留系统迁移
吞吐量对比（10000 并发）：
  R2DBC + WebFlux  | ████████████████████ 12,500 req/s
  JPA + Tomcat     | ████████████          7,200 req/s
  MyBatis + Tomcat | ██████████████        8,500 req/s

9. 常见问题与最佳实践 ​

连接泄漏 ​

java
// 错误：直接创建 DatabaseClient 不使用连接池
DatabaseClient client = DatabaseClient.create(factory);

// 正确：使用连接池 + 超时设置
ConnectionPool pool = new ConnectionPool(
    ConnectionPoolConfiguration.builder(factory)
        .maxAcquireTime(Duration.ofSeconds(5))
        .maxLifeTime(Duration.ofHours(1)).build());
DatabaseClient client = DatabaseClient.builder().connectionFactory(pool).build();

避免事务内阻塞 ​

java
// 错误：混入阻塞调用
@Transactional
public Mono<?> bad() {
    String r = restTemplate.getForObject("http://other/api", String.class);  // 阻塞！
    return repository.save(comment);
}

// 正确：使用 WebClient 保持非阻塞
@Transactional
public Mono<?> good() {
    return webClient.get().uri("http://other/api").retrieve().bodyToMono(String.class)
        .then(repository.save(comment));
}

分页策略 ​

java
// 游标分页（推荐用于弹幕等实时场景）
@Query("SELECT * FROM bullet_comments WHERE video_id = :videoId AND id > :cursor ORDER BY id ASC LIMIT :limit")
Flux<BulletComment> findByCursor(@Param("videoId") String videoId,
                                 @Param("cursor") Long cursor,
                                 @Param("limit") int limit);

// 偏移分页（适合管理后台）
@Query("SELECT * FROM bullet_comments WHERE video_id = :videoId ORDER BY created_at DESC LIMIT :size OFFSET :offset")
Flux<BulletComment> findByPage(@Param("videoId") String videoId,
                               @Param("size") int size,
                               @Param("offset") int offset);

集成测试 ​

java
@Testcontainers
@SpringBootTest
class BulletCommentRepositoryTest {

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15");

    @DynamicPropertySource
    static void configure(DynamicPropertyRegistry reg) {
        reg.add("spring.r2dbc.url", () ->
            "r2dbc:postgresql://" + postgres.getHost() + ":" + postgres.getFirstMappedPort()
            + "/" + postgres.getDatabaseName());
        reg.add("spring.r2dbc.username", postgres::getUsername);
        reg.add("spring.r2dbc.password", postgres::getPassword);
    }

    @Autowired
    private BulletCommentRepository repository;

    @Test
    void testCrud() {
        BulletComment c = new BulletComment();
        c.setUserId("u001"); c.setNickname("小明");
        c.setContent("来了来了！"); c.setColor(0xFFFFFF);
        c.setFontSize(28); c.setPosition(12.5); c.setVideoId("v001");

        StepVerifier.create(repository.save(c).flatMapMany(saved -> repository.findByVideoId("v001")))
            .expectNextMatches(cmt -> cmt.getContent().equals("来了来了！"))
            .verifyComplete();
    }
}

10. 总结 ​

R2DBC 为 Spring WebFlux 提供了原生的响应式数据库访问能力，核心要点：

决策点结论
何时必用全栈响应式架构、高吞吐实时系统（直播弹幕、消息推送、IoT）
何时慎用复杂事务、多表关联查询、团队不熟悉 Reactor
推荐实践DatabaseClient 手写 SQL 用于复杂查询；Repository 用于简单 CRUD
性能关键连接池调优、避免事务内阻塞、游标分页替代 OFFSET 分页
迁移成本从 JPA 迁移需重写 DAO 层，Service 层可复用（保持 Mono/Flux 签名）

R2DBC 不是 JDBC 的替代品，而是响应式领域的补充。在微服务架构中，建议对高吞吐服务使用 R2DBC，对 CRUD 密集的管理后台使用 JPA/MyBatis，实现 IO 模型的最优解耦。