99 Ngày Spring — Ngày 82: Reactive Operators và Backpressure

Code reactive là một chuỗi operator: mỗi operator nhận một publisher và trả về một publisher mới. Dùng sai operator thì chương trình vẫn chạy nhưng sai thứ tự, nuốt lỗi hoặc ngốn bộ nhớ. Bài này gom những operator dùng hằng ngày của Project Reactor và giải thích backpressure, cơ chế để bên nhận nói với bên phát rằng nó chỉ xử lý kịp chừng này.

1. map và flatMap

map biến đổi đồng bộ từng phần tử: hàm trả về giá trị thường. flatMap dùng khi hàm biến đổi tự nó bất đồng bộ và trả về Mono hoặc Flux; operator sẽ subscribe các publisher con và gộp kết quả. Vì các publisher con chạy đồng thời, thứ tự đầu ra của flatMap trên Flux không được bảo đảm. Cần giữ thứ tự và chạy tuần tự thì dùng concatMap; cần chạy đồng thời nhưng trả đúng thứ tự thì dùng flatMapSequential.

Flux<OrderView> views = orderIds
    .flatMap(id -> orders.findById(id), 8)  // async, at most 8 in flight
    .map(OrderView::from);                  // sync transform

Flux<Receipt> ordered = orderIds
    .concatMap(id -> billing.charge(id));   // one at a time, in order

2. zip: ghép nhiều nguồn

Mono.zip subscribe nhiều Mono cùng lúc và phát một tuple khi tất cả đã có giá trị. Đây là cách gọi song song hai ba service rồi ghép kết quả. Nếu một nguồn lỗi, cả zip lỗi; nếu một nguồn hoàn tất rỗng, zip cũng hoàn tất rỗng, nên nguồn không bắt buộc cần giá trị mặc định qua defaultIfEmpty hoặc onErrorReturn.

Mono<ProductPage> page = Mono.zip(
        catalog.find(sku),
        pricing.price(sku),
        reviews.summary(sku).onErrorReturn(ReviewSummary.empty()))
    .map(t -> new ProductPage(t.getT1(), t.getT2(), t.getT3()));

3. Lỗi là một tín hiệu

Trong chuỗi reactive, exception không được ném lên stack mà đi xuôi dòng như một tín hiệu kết thúc. try/catch quanh lời gọi tạo chuỗi không bắt được lỗi xảy ra sau khi subscribe. Dùng onErrorResume để chuyển sang publisher dự phòng, onErrorReturn cho giá trị tĩnh, onErrorMap để đổi loại lỗi. retryWhen với backoff và jitter chỉ dành cho lỗi tạm thời, và luôn đi cùng timeout.

Mono<Price> price = client.get().uri("/prices/{sku}", sku)
    .retrieve()
    .bodyToMono(Price.class)
    .timeout(Duration.ofSeconds(2))
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
        .filter(error -> error instanceof TimeoutException))
    .onErrorResume(error -> cache.lastKnown(sku));

4. Backpressure: bên nhận quyết định tốc độ

Theo đặc tả Reactive Streams, subscriber gọi request(n) để báo nó sẵn sàng nhận n phần tử; publisher không được phát nhiều hơn. Phần lớn operator tự lo việc này. Vấn đề xuất hiện khi nguồn không dừng được, ví dụ sự kiện từ bên ngoài: lúc đó phải chọn chiến lược rõ ràng. Qua HTTP, backpressure dựa vào kiểm soát luồng của TCP: client đọc chậm thì server ghi chậm lại.

• limitRate(n): xin dữ liệu theo từng lô n phần tử.

• onBackpressureBuffer(size): đệm có giới hạn, tràn thì báo lỗi.

• onBackpressureDrop / onBackpressureLatest: bỏ bớt khi dữ liệu mới quan trọng hơn dữ liệu đủ.

• Kiểm thử bằng StepVerifier của reactor-test, kể cả nhánh lỗi.

Flux<Tick> ticks = marketFeed.ticks()
    .onBackpressureLatest()   // a slow client only needs the newest price
    .limitRate(64);

Chọn operator theo ba câu hỏi: đồng bộ hay bất đồng bộ, có cần thứ tự không, và lỗi thì đi đâu. Ngày 83 đưa reactive xuống tầng dữ liệu với R2DBC.

Tài liệu đối chiếu

• https://projectreactor.io/docs/core/release/reference/

• https://www.reactive-streams.org/