Event loop & non-blocking I/O

Bài trước ta có ý tưởng "không thread nào đứng chờ I/O". Bài này xem ai theo dõi các I/O đang chờ: đó là event loop.

1. Bài toán: 1.000 checkout cùng chờ payment

shop-app nhận 1.000 request checkout. Mỗi cái chờ payment-service 2s. Với blocking I/O:

Thread-1   socket.read()  ... ngủ chờ 2s
Thread-2   socket.read()  ... ngủ chờ 2s
...
Thread-1000 socket.read() ... ngủ chờ 2s   => cần 1.000 threads

2. Vấn đề

• Mỗi thread chỉ chờ một socket. Muốn chờ 1.000 socket cần 1.000 thread.
• Thread tốn bộ nhớ, context switch tốn CPU. Tomcat chỉ có 200.

3. Giải pháp: một thread theo dõi nhiều socket

OS có cơ chế (epoll trên Linux, kqueue trên macOS) để hỏi: "trong 1.000 socket này, cái nào đã có dữ liệu?". Java bọc cơ chế này trong java.nio.channels.Selector. Event loop là một vòng lặp vô hạn chạy trên một thread:

while (true) {
    events = selector.select();     // hỏi OS: socket nào sẵn sàng?
    for (event : events) {
        handle(event);              // chạy callback, PHẢI nhanh
    }
    runQueuedTasks();
}

Netty là framework network xây trên NIO. Nó có EventLoopGroup gồm nhiều event loop, mỗi loop là một thread, mỗi connection được gán cố định cho một loop. Reactor Netty là Netty + Project Reactor, và là server mặc định của Spring WebFlux.

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-webflux (đã kèm Reactor Netty).

Bước 2: Config (tùy chọn, thường để mặc định):

# Số event loop worker của Reactor Netty (system property)
java -Dreactor.netty.ioWorkerCount=8 -jar shop-app.jar

Bước 3: Code Java: in tên thread để thấy event loop.

@RestController
public class ThreadDemoController {

    private static final Logger log = LoggerFactory.getLogger(ThreadDemoController.class);
    private final WebClient webClient;

    public ThreadDemoController(WebClient.Builder builder) {
        this.webClient = builder.baseUrl("http://payment-service").build();
    }

    @GetMapping("/demo/payment")
    public Mono<String> payment() {
        log.info("Nhận request trên {}", Thread.currentThread().getName());
        return webClient.get().uri("/payments/status")
                .retrieve().bodyToMono(String.class)
                .doOnNext(r -> log.info("Có kết quả trên {}", Thread.currentThread().getName()));
    }
}

Bước 4: Test

curl localhost:8080/demo/payment
# log:
# Nhận request trên reactor-http-nio-3
# Có kết quả trên reactor-http-nio-3

Thread tên reactor-http-nio-N chính là event loop. Trong 2s chờ payment, nio-3 đi phục vụ connection khác.

5. Cách hoạt động

Event Loop (reactor-http-nio-3)
 |
 |-- P1: gửi HTTP tới payment, đăng ký "khi có response" --> đi tiếp
 |-- P2: gửi HTTP tới payment, đăng ký                   --> đi tiếp
 |-- P3: gửi HTTP tới payment, đăng ký                   --> đi tiếp
 |-- select() ... chờ OS báo
 |
 |<-- OS: socket P2 có data
 |-- chạy callback P2, ghi response cho client
 |<-- OS: socket P1 có data
 |-- chạy callback P1 ...

6. So sánh

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

• Nên: nhiều kết nối, phần lớn thời gian chờ I/O, có client non-blocking (WebClient, R2DBC, reactive Redis/Kafka).
• Không nên: phải gọi JDBC, SDK blocking khắp nơi, hoặc xử lý nặng CPU trong callback.

Đúng rồi, bạn hình dung khá sát. Chỉ cần chỉnh 2 điểm:

1. Nó không đứng chờ từng I/O. Nó gọi selector.select() để chờ bất kỳ socket nào sẵn sàng. Một thread theo dõi hàng nghìn connection, không bị giữ riêng cho P1, P2 hay P3.
2. Không nhất thiết kéo worker thread. Thường chính event loop chạy luôn callback (parse JSON, map dữ liệu, ghi response). Chỉ khi có việc blocking bắt buộc thì mới đẩy sang worker pool như Schedulers.boundedElastic().

Event Loop Thread
 | nhận request
 | việc nhẹ: tự làm luôn
 | việc blocking bắt buộc (JDBC, SDK cũ):
 |------------------------------> boundedElastic worker
 |   rảnh, xử lý I/O khác           | gọi JDBC, BLOCK 2s
 |<------------------------------ | có kết quả
 | tiếp tục trả response

@GetMapping("/legacy-payment")
public Mono<String> legacy() {
    return Mono.fromCallable(() -> legacyBankSdk.pay())      // blocking
               .subscribeOn(Schedulers.boundedElastic());   // chạy trên worker
}

Chính xác cả hai ý. Đây là rủi ro lớn nhất của event loop. Giả sử có 2 loop, mỗi loop 500 connection, và request P1 chạy Thread.sleep(5000) trên event-loop-1:

event-loop-1  [BLOCKED 5s] P1 sleep    P2..P500: chờ, dù chúng chẳng liên quan
event-loop-2  [RUNNING]    P501..P1000: bình thường

Một bug chặn 500 người dùng, không phải 1. Block hết các loop là server gần như đứng.

Lỗi hay gặp nhất, trông vô hại:

// SAI: callBankBlocking() chạy NGAY khi tạo Mono, trên event loop
return Mono.just(paymentService.callBankBlocking());

// ĐÚNG (nếu buộc phải blocking): lazy + đẩy sang worker
return Mono.fromCallable(paymentService::callBankBlocking)
           .subscribeOn(Schedulers.boundedElastic());

Production tránh bằng cách:

• Không dùng Thread.sleep(), JDBC, Future.get(), .block() trên event loop.
• Dùng WebClient, R2DBC. Việc blocking bắt buộc đưa sang boundedElastic.
• Bật BlockHound khi test để phát hiện blocking call trên thread non-blocking.
• Lưu ý: tính toán CPU 5s trong callback cũng chiếm loop y như blocking I/O.

Đúng rồi, WebFlux (Netty) và FastAPI (Uvicorn/asyncio) cùng tư duy. Còn số lượng: không phải càng nhiều càng tốt. Reactor Netty mặc định:

ioWorkerCount = max(4, số CPU khả dụng)

Event loop không chờ I/O, nên thread nhiều hơn core chỉ thêm context switch. 8 loop vẫn phục vụ hàng nghìn connection nếu request chủ yếu chờ I/O; nếu mỗi request tốn 100ms CPU thì 8 loop nhanh chóng thành giới hạn. Production: giữ mặc định, load test, xem CPU, p95/p99, event loop latency. Hết sức thì scale ngang:

          Load Balancer
       /       |       \
  Instance1 Instance2 Instance3
  8 loops   8 loops   8 loops

Q1. Event loop là gì?

🇻🇳 Tiếng Việt:

Event loop là một thread chạy vòng lặp: hỏi OS qua selector xem socket nào sẵn sàng rồi chạy callback tương ứng. Vì không đứng chờ từng I/O, một thread quản lý được hàng nghìn kết nối. Netty có EventLoopGroup gồm nhiều loop, Spring WebFlux chạy trên Reactor Netty.

🇬🇧 English:

An event loop is a thread running a loop: it asks the OS through a selector which sockets are ready and then runs the matching callbacks. Since it never waits on a single I/O operation, one thread can manage thousands of connections. Netty uses an EventLoopGroup of several loops, and Spring WebFlux runs on Reactor Netty.

Q2. Chuyện gì xảy ra nếu event loop bị block?

🇻🇳 Tiếng Việt:

Mọi connection gán cho loop đó bị chậm, không chỉ request gây lỗi. Vì vậy không gọi JDBC, sleep hay block() trên event loop. Việc blocking bắt buộc thì bọc fromCallable và subscribeOn boundedElastic, và dùng BlockHound khi test.

🇬🇧 English:

Every connection assigned to that loop stalls, not just the request that caused it. So we never call JDBC, sleep or block() on an event loop. Unavoidable blocking work is wrapped in fromCallable and moved to boundedElastic with subscribeOn, and we use BlockHound in tests.

Q3. Cần bao nhiêu event loop thread?

🇻🇳 Tiếng Việt:

Khoảng bằng số CPU core; Reactor Netty mặc định là max(4, số processor). Thêm nhiều hơn không giúp vì loop không chờ I/O. Bắt đầu từ mặc định, load test, rồi scale ngang thêm instance khi cần.

🇬🇧 English:

Roughly the number of CPU cores; Reactor Netty defaults to max(4, available processors). More threads do not help because loops never wait on I/O. Start with the default, load test, and scale out horizontally when needed.

Cách nhớ ngắn gọn

• Event loop = 1 thread + selector + callback nhanh.
• Số loop ≈ số core. Hết sức thì scale ngang.
• Không bao giờ block loop; blocking bắt buộc → boundedElastic.
• Mono.just(blocking()) là bẫy: chạy ngay trên loop.
• Giống FastAPI: time.sleep trong async def = Thread.sleep trong WebFlux.