Module 07 · Executable

Project Reactor

Reactive composition với Mono/Flux, operators, scheduler, backpressure, testing và debugging.

1. Mono và Flux

TypeCardinalityVí dụ
Mono<T>0..1Find user by id.
Flux<T>0..NRows, events, stream.

2. Creation

Mono.just("hello");
Mono.empty();
Mono.error(new IllegalStateException());

Flux.just(1, 2, 3);
Flux.range(1, 10);
Flux.fromIterable(items);

Mono.fromCallable(this::blockingCall);
Mono.defer(() -> computePublisherAtSubscriptionTime());

defer tạo publisher tại thời điểm subscribe, hữu ích khi cần state mới cho mỗi subscription.

3. map / flatMap / concatMap

Mono<String> name = userMono.map(User::name);

Mono<Order> order =
    userMono.flatMap(user -> orderRepository.findLatest(user.id()));
Bridge: thenCompose ≈ flatMap ≈ Vert.x compose.
  • flatMap: concurrency cao, completion order có thể khác source order.
  • concatMap: tuần tự, giữ thứ tự.
  • flatMapSequential: inner có thể concurrent nhưng output giữ source order.

4. zip / merge / concat

Mono<Profile> profile =
    Mono.zip(userMono, ordersMono)
        .map(tuple -> new Profile(tuple.getT1(), tuple.getT2()));

merge interleave theo arrival time; concat chờ source trước hoàn tất.

5. Error handling

serviceCall()
    .timeout(Duration.ofSeconds(2))
    .retryWhen(Retry.backoff(2, Duration.ofMillis(100)))
    .onErrorResume(TimeoutException.class, ex -> fallback());
Retry side effect phải xét idempotency.

6. Scheduler

SchedulerDùng khi
parallel()CPU work ngắn.
boundedElastic()Bridge blocking APIs có giới hạn.
single()Serial execution.
source
    .subscribeOn(Schedulers.boundedElastic())
    .map(this::stepA)
    .publishOn(Schedulers.parallel())
    .map(this::cpuStep);

subscribeOn tác động subscription/upstream; publishOn tạo execution boundary cho downstream.

7. Backpressure

flux.limitRate(100);
flux.onBackpressureBuffer(1000);
flux.onBackpressureDrop();
flux.onBackpressureLatest();

8. Cold vs Hot

Cold: mỗi subscriber có sequence riêng. Hot: subscribers quan sát nguồn đang phát theo thời gian.

9. StepVerifier

StepVerifier.create(service.findUser(42))
    .expectNextMatches(user -> user.id() == 42)
    .verifyComplete();

10. Debugging

pipeline
    .checkpoint("load-order")
    .doOnNext(v -> log.debug("value={}", v))
    .doOnError(e -> log.error("pipeline failed", e));

11. Lab — Aggregator

findUser(id)   ┐
               ├─ zip ─► profile
findOrders(id) ┘
        ↓
timeout + selective retry + fallback orders=[]

Không dùng block(); test timeout bằng StepVerifier.

12. Bài tập

G1. Khi nào map tạo nested publisher?
Khi function trả Publisher; dùng flatMap.
G2. Chọn flatMap hay concatMap nếu bắt buộc đúng thứ tự?
G3. Vì sao JDBC trong map trên event-loop thread là anti-pattern?

Tài liệu và code thực hành

Đọc chapter chi tiết [legacy source: modules/07-project-reactor.md] · Mở foundation labs [legacy source: labs/foundation/README.md] · Xem Reactor source [legacy source: examples/README.md]

← Module 06Module 08 — WebFlux →