Backpressure: khi producer nhanh hơn consumer

Đúng, đây là thứ mà callback hay CompletableFuture không có. Backpressure = consumer nói cho producer biết nó nhận được bao nhiêu.

1. Bài toán: đối soát đơn hàng cuối ngày

shop-app có job đối soát: đọc 1 triệu đơn hàng từ DB (R2DBC), mỗi đơn gọi payment-service để kiểm tra trạng thái. DB đọc 50.000 dòng/giây, payment-service chỉ chịu 100 request/giây.

DB (producer)       ──── 50.000/s ────>   gọi payment (consumer) 100/s

2. Vấn đề

Nếu producer cứ đẩy không quan tâm consumer:

đơn chờ xử lý trong RAM: 49.900 thêm mỗi giây
sau 20s: ~1 triệu object trong heap  ->  OutOfMemoryError
hoặc bắn 50.000 req/s vào payment-service  ->  sập payment

Với Kafka/queue thì message nằm trên đĩa. Còn ở đây mọi thứ nằm trong bộ nhớ JVM.

3. Giải pháp: pull theo nhu cầu

Reactive Streams biến luồng thành push-pull: Subscriber gọi subscription.request(n), Publisher chỉ được phát tối đa n phần tử cho tới lần request tiếp theo.

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-webflux, spring-boot-starter-data-r2dbc + driver (ví dụ r2dbc-postgresql).

Bước 2: Config (mật khẩu lấy từ biến môi trường, không hardcode):

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/shop
    username: ${DB_USER}
    password: ${DB_PASSWORD}

Bước 3: Code Java. Thấy request(n) bằng tay:

Flux.range(1, 10)
    .log()                                   // in ra request(n), onNext...
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription s) {
            request(2);                      // chỉ xin 2
        }
        @Override
        protected void hookOnNext(Integer v) {
            System.out.println("Xử lý " + v);
            request(1);                      // xong 1 cái xin thêm 1
        }
    });

| request(2)
| onNext(1)  Xử lý 1  | request(1)
| onNext(2)  Xử lý 2  | request(1)
...

Job đối soát thực tế, không cần viết Subscriber tay:

@Service
public class ReconcileJob {

    private final OrderRepository orderRepository;   // ReactiveCrudRepository
    private final PaymentClient paymentClient;

    public Mono<Long> run(LocalDate day) {
        return orderRepository.findByCreatedDate(day)        // Flux<Order>, 1 triệu dòng
                .limitRate(500)                              // xin DB từng lô 500
                .flatMap(o -> paymentClient.status(o.id())   // tối đa 20 call song song
                                  .map(s -> check(o, s)), 20)
                .filter(Result::mismatch)
                .concatMap(r -> orderRepository.markMismatch(r.orderId()))
                .count();
    }
}

Nguồn không hiểu backpressure (ví dụ sự kiện từ listener ngoài, Flux.interval):

// Giá sản phẩm cập nhật liên tục, UI chỉ cần giá mới nhất
Flux<Price> prices = priceTicker.stream()     // phát mỗi 10ms
        .onBackpressureLatest();

// Log click chuột: mất vài cái cũng được
Flux<Click> clicks = clickSource.stream()
        .onBackpressureDrop(c -> metrics.increment("click.dropped"));

// Đơn hàng: không được mất, đệm có giới hạn, tràn thì báo lỗi
Flux<OrderEvent> events = orderSource.stream()
        .onBackpressureBuffer(10_000);

Bước 4: Test

StepVerifier.create(Flux.range(1, 100), 0)   // ban đầu request 0
    .thenRequest(3)
    .expectNext(1, 2, 3)
    .thenCancel()
    .verify();

curl -X POST localhost:8080/admin/reconcile?day=2026-10-13
# theo dõi: heap ổn định, payment-service nhận ~20 request song song

5. Cách hoạt động

Subscriber                     limitRate(500)             R2DBC (DB)
   | -- request(20) --> flatMap |                           |
   |                            | -- request(500) ------->  |
   |                            | <-- 500 rows ------------ |
   |   flatMap chạy 20 call song song
   |   xong 1 call -> request(1) thêm
   |                            | đã dùng 375 (75%) -> request(375)
   |                            | <-- 375 rows ------------ |

Producer luôn chạy chậm theo consumer. RAM ~ vài trăm đơn, không phải 1 triệu.

Với HTTP, backpressure còn đi xuống tầng TCP: client đọc chậm thì Netty ngừng đọc socket, TCP window đầy, phía gửi phải chậm lại.

6. So sánh

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

• Xử lý Flux lớn: luôn giới hạn concurrency (flatMap(fn, n)) và lô (limitRate).
• onBackpressureBuffer() không giới hạn chỉ dời OOM ra sau. Luôn đặt max.
• Mono chỉ 1 phần tử, backpressure gần như không liên quan.
• Backpressure chỉ trong một JVM/kết nối. Giữa các service lâu dài thì dùng message broker (Kafka) và consumer lag.

Không, chỉ cần chỉnh một điểm. Có hai loại nguồn:

• Nguồn pull được (DB cursor, Flux.range, list): chờ request rồi mới đọc tiếp. Backpressure hoạt động hoàn hảo.
• Nguồn theo thời gian / sự kiện ngoài (interval, click, sensor): thời gian không dừng được. Consumer không request kịp thì Flux.interval ném OverflowException ("Could not emit tick ... due to lack of requests").

Với loại thứ hai bạn phải chọn chiến lược: buffer, drop hay latest. Đó chính là lúc dùng các operator onBackpressureXxx.

Flux.interval(Duration.ofMillis(1))
    .onBackpressureDrop()
    .concatMap(t -> slowSave(t), 1)
    .subscribe();

Gần đúng, cùng mục đích bảo vệ phía chậm, nhưng khác cơ chế:

Trong job đối soát, flatMap(fn, 20) là backpressure. Còn nếu payment-service quy định "tối đa 100 req/s" thì thêm rate limit (ví dụ Resilience4j RateLimiter) ở phía gọi.

Q1. Backpressure là gì?

🇻🇳 Tiếng Việt:

Là cơ chế để consumer báo cho producer biết nó xử lý được bao nhiêu, tránh bị dồn dữ liệu gây OOM hoặc quá tải. Trong Reactive Streams, subscriber gọi subscription.request(n) và publisher chỉ phát tối đa n phần tử. Reactor có limitRate và tham số concurrency của flatMap để điều chỉnh.

🇬🇧 English:

It is the mechanism that lets a consumer tell the producer how much it can handle, so it is not flooded and does not run out of memory. In Reactive Streams the subscriber calls subscription.request(n) and the publisher emits at most n items. Reactor offers limitRate and the concurrency argument of flatMap to tune it.

Q2. Nguồn không hỗ trợ backpressure thì làm sao?

🇻🇳 Tiếng Việt:

Với nguồn theo thời gian hoặc sự kiện ngoài, phải chọn chiến lược: onBackpressureBuffer có giới hạn nếu không được mất dữ liệu, onBackpressureDrop nếu mất được, onBackpressureLatest nếu chỉ cần giá trị mới nhất. Buffer không giới hạn chỉ dời OOM ra sau.

🇬🇧 English:

For time-based or external event sources we must pick a strategy: a bounded onBackpressureBuffer when data must not be lost, onBackpressureDrop when losing items is acceptable, and onBackpressureLatest when only the newest value matters. An unbounded buffer just postpones the OutOfMemoryError.

Q3. Xử lý một triệu bản ghi bằng Flux mà không làm sập downstream?

🇻🇳 Tiếng Việt:

Stream bằng R2DBC thay vì load hết, dùng limitRate để đọc theo lô, flatMap với concurrency giới hạn khi gọi service khác, concatMap khi cần thứ tự. Bộ nhớ chỉ giữ vài trăm phần tử, và producer tự chạy chậm theo consumer.

🇬🇧 English:

Stream the rows with R2DBC instead of loading them all, use limitRate to fetch in batches, flatMap with bounded concurrency when calling other services, and concatMap when order matters. Memory holds only a few hundred items, and the producer slows down to match the consumer.

Cách nhớ ngắn gọn

• Backpressure = consumer request(n), producer không phát quá n.
• limitRate = lô; flatMap(fn, n) = giới hạn song song.
• Nguồn không pull được: Buffer (có max) / Drop / Latest.
• Backpressure ≠ rate limiting.