Schedulers: code reactive chạy trên thread nào?

Câu hỏi đúng lúc. Bài này trả lời hai chuyện: một pipeline Mono/Flux chạy trên thread nào, và cách đẩy việc blocking ra khỏi event loop bằng Schedulers.

1. Bài toán: checkout gọi một SDK cũ

Project shop-app (giống chương Concurrency) đã chuyển sang WebFlux. API checkout cần:

POST /checkout/{userId}
  |-- userClient.get(userId)          (WebClient, non-blocking)
  |-- orderClient.getCart(userId)     (WebClient, non-blocking)
  |-- legacyTaxSdk.calculate(cart)    (thư viện cũ, BLOCKING ~300ms)

Thư viện thuế là JAR của bên thứ ba, chỉ có API đồng bộ. Không sửa được.

2. Vấn đề: gọi thẳng là chặn event loop

return cartMono.map(cart -> legacyTaxSdk.calculate(cart)); // ❌ chạy trên reactor-http-nio-*

reactor-http-nio-2 (1 trong ~số-core thread)
  |-- request A: calculate() ===== BLOCK 300ms =====
  |-- request B, C, D ... đứng chờ cùng thread này

Event loop chỉ có khoảng số core CPU thread. Mỗi lần block là hàng nghìn connection bị treo theo.

3. Giải pháp: Schedulers

Mặc định, Reactor không tự đổi thread. Code chạy trên thread đã gọi subscribe() hoặc thread phát ra tín hiệu (với WebFlux là event loop). Muốn đổi thread thì dùng Scheduler:

Hai operator để chuyển thread:

• subscribeOn(s): đổi thread nơi bắt đầu subscribe, tức là nguồn phát dữ liệu. Đặt ở đâu trong chuỗi cũng tác động lên nguồn. Nhiều subscribeOn thì cái gần nguồn nhất thắng.
• publishOn(s): đổi thread cho các operator phía sau nó. Vị trí quan trọng.

4. Code từng bước

Bước 1: dependency (đã có từ các bài trước)

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Bước 2: bọc code blocking đúng cách

@Service
@RequiredArgsConstructor
public class TaxService {
    private final LegacyTaxSdk legacyTaxSdk;

    public Mono<BigDecimal> calculate(Cart cart) {
        return Mono.fromCallable(() -> legacyTaxSdk.calculate(cart)) // lazy: chỉ chạy khi subscribe
                   .subscribeOn(Schedulers.boundedElastic())          // chạy trên thread được phép block
                   .timeout(Duration.ofSeconds(2));
    }
}

Chú ý dùng fromCallable, không dùng Mono.just(legacyTaxSdk.calculate(cart)). just tính giá trị ngay lúc tạo, trên thread hiện tại, nên subscribeOn vô tác dụng.

Bước 3: ghép vào checkout

public Mono<CheckoutResult> checkout(Long userId) {
    return orderClient.getCart(userId)                 // event loop
        .flatMap(cart -> taxService.calculate(cart)    // boundedElastic
            .map(tax -> new CheckoutResult(cart, tax)))
        .publishOn(Schedulers.parallel())              // phía sau chạy trên parallel
        .map(this::signReceipt);                       // CPU-bound: ký số hóa đơn
}

Bước 4: in tên thread để kiểm chứng

Flux.range(1, 2)
    .doOnNext(i -> log("A", i))
    .publishOn(Schedulers.parallel())
    .doOnNext(i -> log("B", i))
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe(i -> log("C", i));

// A 1 boundedElastic-1   <- nguồn chạy theo subscribeOn
// B 1 parallel-1         <- sau publishOn
// C 1 parallel-1

curl -X POST localhost:8080/checkout/42
# log: TaxService chạy trên boundedElastic-3, controller trên reactor-http-nio-2

5. Cách hoạt động

reactor-http-nio-2         boundedElastic-3            parallel-1
     |                            |                           |
 nhận request                     |                           |
 getCart() (non-blocking I/O)     |                           |
     |--- subscribeOn ----------> calculate() BLOCK 300ms     |
 (rảnh, phục vụ request khác)     |                           |
     |                            |--- publishOn -----------> signReceipt()
     |<------------------------- ghi response -----------------|

6. So sánh

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

• Nên: bọc vài lời gọi blocking không tránh được (SDK cũ, file, JDBC tạm thời).
• Không nên: bọc toàn bộ JPA bằng boundedElastic rồi gọi là "reactive". Lúc đó bạn đang quay lại mô hình thread-per-request, chỉ phức tạp hơn. Hãy cân nhắc MVC + virtual threads (bài 4.5.10).
• Không dùng parallel() cho việc blocking: nó chỉ có số-core thread, block là nghẽn y như event loop.

Có: BlockHound. Nó là Java agent, cài vào lúc chạy test, và ném lỗi khi một thread "non-blocking" (event loop, parallel) gọi hàm blocking như Thread.sleep, đọc socket, FileInputStream.read.

<dependency>
  <groupId>io.projectreactor.tools</groupId>
  <artifactId>blockhound</artifactId>
  <scope>test</scope>
</dependency>

class BlockingDetectionTest {
    @BeforeAll
    static void setUp() { BlockHound.install(); }

    @Test
    void blockingOnParallelIsDetected() {
        Mono<String> bad = Mono.fromCallable(() -> { Thread.sleep(10); return "x"; })
                               .subscribeOn(Schedulers.parallel());
        StepVerifier.create(bad)
            .expectErrorMatches(e -> e.getMessage().contains("Blocking call"))
            .verify();
    }
}

Thread boundedElastic được BlockHound cho phép block, nên đổi sang boundedElastic() là hết lỗi. Trên JDK 13+ cần thêm JVM option -XX:+AllowRedefinitionToAddDeleteMethods khi chạy test.

Chỉ bật trong test/staging, không bật production vì có overhead instrument bytecode.

Đúng rồi, và đây là chỗ hay nhầm nhất. Nhớ rằng Reactor lazy: tín hiệu subscribe đi ngược từ dưới lên, còn dữ liệu chảy từ trên xuống.

subscribe() ──> subscribeOn ──> map ──> fromCallable   (subscribe đi lên)
                  | đổi thread ở đây, nên mọi thứ phía trên chạy trên boundedElastic

fromCallable ──> map ──> publishOn ──> doOnNext       (dữ liệu đi xuống)
                           | đổi thread ở đây, chỉ ảnh hưởng phía dưới

Hình dung: subscribeOn là "chọn nhà máy nơi bắt đầu sản xuất", còn publishOn là "trạm trung chuyển, từ đây hàng lên xe khác".

Q1. subscribeOn và publishOn khác nhau thế nào?

🇻🇳 Tiếng Việt:

subscribeOn đổi thread nơi nguồn được subscribe và phát dữ liệu, đặt ở đâu trong chuỗi cũng được, nếu có nhiều cái thì cái gần nguồn nhất thắng. publishOn đổi thread cho các operator đứng sau nó, nên vị trí quan trọng. Em dùng subscribeOn(boundedElastic) để bọc nguồn blocking và publishOn(parallel) để chuyển xử lý CPU nặng.

🇬🇧 English:

subscribeOn changes the thread on which the source is subscribed and emits, so its position in the chain does not matter, and the one closest to the source wins. publishOn switches the thread for the operators after it, so its position matters. I use subscribeOn with boundedElastic to wrap a blocking source, and publishOn with parallel to move CPU-heavy work.

Q2. Làm sao gọi code blocking trong ứng dụng WebFlux?

🇻🇳 Tiếng Việt:

Bọc bằng Mono.fromCallable để lazy, rồi subscribeOn(Schedulers.boundedElastic()), thêm timeout. Không dùng Mono.just vì nó chạy ngay trên event loop. Nếu phần lớn code là blocking thì nên xem lại lựa chọn, có thể dùng Spring MVC với virtual threads.

🇬🇧 English:

Wrap it in Mono.fromCallable so it is lazy, then subscribeOn Schedulers.boundedElastic and add a timeout. Never use Mono.just, because it evaluates immediately on the event loop. If most of the code is blocking, I would reconsider the stack, for example Spring MVC with virtual threads.

Q3. Schedulers.parallel và boundedElastic khác nhau ra sao?

🇻🇳 Tiếng Việt:

parallel có số thread bằng số core, dành cho việc CPU-bound ngắn, không được block. boundedElastic có giới hạn thread (mặc định 10 lần số core) và hàng đợi, dành cho việc blocking. Để phát hiện block nhầm chỗ, em dùng BlockHound trong test.

🇬🇧 English:

parallel has one thread per CPU core and is for short CPU-bound work that must never block. boundedElastic has a capped number of threads, ten times the cores by default, plus a task queue, and it is meant for blocking work. To catch accidental blocking I run BlockHound in tests.

Cách nhớ ngắn gọn

• Reactor không tự đổi thread; Scheduler mới đổi.
• subscribeOn = nguồn, vị trí không quan trọng. publishOn = phía sau, vị trí quan trọng.
• Blocking → fromCallable + boundedElastic. CPU → parallel.
• BlockHound bắt lỗi block event loop trong test.