R2DBC: database cũng phải non-blocking

Vì JPA đứng trên JDBC, mà JDBC là blocking từ gốc. Bài này giải thích vì sao, và cách thay bằng R2DBC.

1. Bài toán: lưu đơn hàng khi checkout

shop-app đã có WebFlux + WebClient. Bước cuối của checkout là ghi bảng orders và order_items vào PostgreSQL, rồi trả về mã đơn.

2. Vấn đề: JDBC block thread

ResultSet rs = statement.executeQuery(sql); // thread đứng chờ socket DB trả dữ liệu

reactor-http-nio-1
  | orderRepository.save(order)  (JPA)
  | ===== BLOCK 50ms chờ PostgreSQL =====
  | mọi request khác trên thread này đứng chờ

• API JDBC (executeQuery, ResultSet.next) trả kết quả trực tiếp, nên bắt buộc chờ.
• JPA thêm lazy loading: truy cập order.getItems() có thể bắn query ẩn, cũng blocking.
• Bọc bằng boundedElastic hay CompletableFuture.supplyAsync chỉ dời chỗ chờ sang worker thread, không làm JDBC non-blocking.

3. Giải pháp: R2DBC

R2DBC (Reactive Relational Database Connectivity) là đặc tả driver SQL non-blocking, API trả Publisher. Có driver cho PostgreSQL, MySQL/MariaDB, SQL Server, Oracle, H2.

Spring Data R2DBC đặt lên trên: repository trả Mono/Flux, DatabaseClient cho SQL tự viết, transaction reactive.

Nó không phải ORM. Không có:

• Lazy loading, dirty checking, persistence context / first-level cache.
• Quan hệ @OneToMany, @ManyToOne, cascade.
• Tự sinh schema (dùng Flyway/Liquibase hoặc schema.sql).

4. Code từng bước

Bước 1: dependency

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
  <groupId>org.postgresql</groupId>
  <artifactId>r2dbc-postgresql</artifactId>
  <scope>runtime</scope>
</dependency>

Bước 2: config (password lấy từ biến môi trường)

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/shop
    username: ${DB_USER}
    password: ${DB_PASSWORD}
    pool:
      initial-size: 5
      max-size: 20

Bước 3: entity và repository

@Table("orders")
public record Order(@Id Long id, Long userId, BigDecimal total, String status) {}

@Table("order_items")
public record OrderItem(@Id Long id, Long orderId, Long productId, int quantity) {}

public interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
    Flux<Order> findByUserId(Long userId);
}
public interface OrderItemRepository extends ReactiveCrudRepository<OrderItem, Long> {
    Flux<OrderItem> findByOrderId(Long orderId);
}

Bước 4: service có transaction reactive

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orders;
    private final OrderItemRepository items;

    @Transactional   // dùng R2dbcTransactionManager, tự cấu hình bởi Boot
    public Mono<Order> place(Long userId, Cart cart) {
        return orders.save(new Order(null, userId, cart.total(), "PENDING"))
            .flatMap(saved -> Flux.fromIterable(cart.lines())
                .map(l -> new OrderItem(null, saved.id(), l.productId(), l.quantity()))
                .concatMap(items::save)        // lỗi ở đây → rollback cả đơn
                .then(Mono.just(saved)));
    }

    // "quan hệ" phải tự ghép
    public Mono<OrderDetail> detail(Long orderId) {
        return orders.findById(orderId)
            .zipWith(items.findByOrderId(orderId).collectList(), OrderDetail::new);
    }
}

Bước 5: test bằng curl

curl -X POST localhost:8080/orders -H "Content-Type: application/json" \
     -d '{"userId":42}'
# {"id":1001,"userId":42,"total":350000,"status":"PENDING"}
curl localhost:8080/orders/1001

5. Cách hoạt động

reactor-http-nio-1                  PostgreSQL
  | orders.save()  -- gửi INSERT -->  |
  | (không chờ, phục vụ request khác) |
  | <-- kết quả đến (NIO event) ------ |
  | flatMap: items.save() -- INSERT -->|
  | ...                                |
  | COMMIT khi Mono complete, ROLLBACK khi có onError

@Transactional reactive: connection + transaction gắn vào Reactor Context
(không phải ThreadLocal, vì pipeline có thể nhảy thread)

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: app đã chọn WebFlux, truy vấn đơn giản, cần nhiều kết nối đồng thời, muốn stream kết quả lớn.
• Không nên: domain phức tạp nhiều quan hệ, team dựa nhiều vào JPA, hoặc cần các tính năng Hibernate. Khi đó MVC + JPA + virtual threads hợp hơn.
• Đừng trộn: WebFlux + JPA gọi thẳng = block event loop.

Đúng rồi, có hai cách, như trong detail() ở trên:

1. Hai query rồi ghép: findById + findByOrderId().collectList(), nối bằng zipWith. Đơn giản, hai query chạy song song.
2. Một JOIN bằng DatabaseClient, rồi tự group:

databaseClient.sql("""
        SELECT o.id, o.status, i.product_id, i.quantity
        FROM orders o JOIN order_items i ON i.order_id = o.id
        WHERE o.id = :id""")
    .bind("id", orderId)
    .map((row, meta) -> new Line(row.get("product_id", Long.class),
                                 row.get("quantity", Integer.class)))
    .all()          // Flux<Line>
    .collectList();

Nhìn thì nhiều code hơn, nhưng bù lại không có N+1 ẩn: query nào chạy là bạn thấy ngay trong code.

Bạn hiểu đúng điểm khó. Spring giải quyết bằng Reactor Context thay cho ThreadLocal: transaction và connection được gắn vào Context của pipeline, đi theo subscription chứ không theo thread.

Hệ quả cần nhớ:

• Method @Transactional phải trả về Mono/Flux và caller phải subscribe; transaction chỉ bắt đầu khi subscribe.
• Nếu bạn tự subscribe() bên trong (fire-and-forget), phần đó nằm ngoài transaction.
• Muốn kiểm soát bằng code, dùng TransactionalOperator:

return orders.save(order)
    .flatMap(o -> items.saveAll(lines(o)).then(Mono.just(o)))
    .as(transactionalOperator::transactional);

Q1. Vì sao không dùng JPA trong ứng dụng WebFlux?

🇻🇳 Tiếng Việt:

JPA chạy trên JDBC, một API blocking: thread phải chờ database trả kết quả. Gọi trên event loop sẽ chặn mọi request khác. Bọc bằng boundedElastic chỉ dời việc chờ sang thread khác. Muốn end-to-end non-blocking thì dùng R2DBC, hoặc nếu cần JPA thì chọn MVC với virtual threads.

🇬🇧 English:

JPA is built on JDBC, which is a blocking API, so the thread waits for the database. Calling it on an event loop stalls every other request. Wrapping it in boundedElastic only moves the waiting to another thread. For end-to-end non-blocking I/O I use R2DBC, or if I really need JPA, I choose Spring MVC with virtual threads.

Q2. Spring Data R2DBC khác JPA ở điểm nào?

🇻🇳 Tiếng Việt:

R2DBC là driver non-blocking, repository trả Mono và Flux. Nó không phải ORM: không lazy loading, không persistence context, không mapping quan hệ hay cascade. Quan hệ phải tự query rồi ghép bằng zip, hoặc viết JOIN qua DatabaseClient. Bù lại query rõ ràng, không có N+1 ẩn.

🇬🇧 English:

R2DBC is a non-blocking driver, and its repositories return Mono and Flux. It is not an ORM: there is no lazy loading, no persistence context, no relationship mapping and no cascade. I load related data with separate queries combined with zip, or write a JOIN with DatabaseClient. The upside is explicit queries with no hidden N+1 problem.

Q3. Transaction trong R2DBC hoạt động thế nào?

🇻🇳 Tiếng Việt:

Dùng @Transactional trên method trả Mono/Flux, với R2dbcTransactionManager, hoặc TransactionalOperator. Transaction gắn vào Reactor Context thay vì ThreadLocal vì pipeline có thể đổi thread. Commit khi pipeline complete, rollback khi có error. Phần nào tự subscribe riêng sẽ nằm ngoài transaction.

🇬🇧 English:

I use @Transactional on methods that return Mono or Flux, backed by R2dbcTransactionManager, or TransactionalOperator. The transaction is bound to the Reactor Context instead of a ThreadLocal, because the pipeline can switch threads. It commits when the pipeline completes and rolls back on error. Anything subscribed separately runs outside the transaction.

Cách nhớ ngắn gọn

• JDBC/JPA = blocking từ gốc; bọc thread khác chỉ dời chỗ chờ.
• R2DBC = driver non-blocking; Spring Data R2DBC = repository Mono/Flux.
• Không ORM: không lazy, không relation, tự ghép bằng zip/JOIN.
• Transaction theo Reactor Context; commit khi complete, rollback khi error.