Async MVC: Callable, DeferredResult, CompletableFuture, SseEmitter

Có: SSE (Server-Sent Events) với SseEmitter. Nhưng SSE là một trường hợp của chuyện lớn hơn: trả response mà không giữ thread Tomcat. Đi từ gốc.

1. Bài toán

• GET /api/orders/{id}/status/stream: khách mở trang, nhận PAID → PACKED → SHIPPED theo thời gian thực.
• GET /api/orders/{id}/payment: chờ payment-service xác nhận, có thể 5–10 giây (webhook về sau).
• GET /api/orders/{id}/summary: gọi song song payment, shipping, invoice.

2. Vấn đề

Tomcat: server.tomcat.threads.max = 200 (platform threads)

1000 khách mở trang theo dõi, mỗi request giữ 1 thread để chờ sự kiện
→ 200 thread bị chiếm hết, request mới (kể cả GET /products) xếp hàng → app "đứng"

Polling 2 giây cũng tốn: 1000 khách = 500 request/giây, đa số trả "chưa đổi". Xem Tomcat threads.

3. Giải pháp

Servlet 3 có async: controller trả một "lời hứa", Spring gọi request.startAsync(), trả thread về pool, giữ connection mở. Khi có kết quả, Spring dispatch lại để ghi response.

• Callable<T>: Spring chạy nó trên AsyncTaskExecutor. Giải phóng thread Tomcat nhưng chiếm thread khác.
• DeferredResult<T>: bạn tự setResult() từ bất kỳ đâu (listener Kafka, webhook). Không chiếm thread khi chờ.
• CompletableFuture<T>: giống DeferredResult, hợp khi đã có future (xem CompletableFuture).
• SseEmitter: gửi nhiều sự kiện trên một response text/event-stream.
• StreamingResponseBody: ghi stream thô (bài 4.2.10).

4. Code từng bước

Bước 1: Dependency. Chỉ spring-boot-starter-web.

Bước 2: Config

spring:
  mvc:
    async:
      request-timeout: 30s        # mặc định cho async request
  task:
    execution:                    # executor cho Callable / StreamingResponseBody
      pool:
        core-size: 16
        max-size: 64
        queue-capacity: 200
      thread-name-prefix: mvc-async-

Bước 3: Code Java

DeferredResult: chờ webhook payment.

@RestController
@RequiredArgsConstructor
public class PaymentStatusController {

    private final Map<Long, DeferredResult<PaymentStatus>> waiting = new ConcurrentHashMap<>();

    @GetMapping("/api/orders/{id}/payment")
    public DeferredResult<PaymentStatus> await(@PathVariable Long id) {
        var result = new DeferredResult<PaymentStatus>(10_000L, PaymentStatus.PENDING); // timeout → PENDING
        waiting.put(id, result);
        result.onCompletion(() -> waiting.remove(id, result));
        return result;                                   // thread Tomcat được trả về ngay
    }

    @PostMapping("/webhooks/payment")                     // payment-service gọi lại
    public ResponseEntity<Void> webhook(@RequestBody PaymentEvent e) {
        var r = waiting.get(e.orderId());
        if (r != null) r.setResult(e.status());          // ghi response cho request đang chờ
        return ResponseEntity.noContent().build();
    }
}

CompletableFuture: gộp 3 lời gọi song song.

@GetMapping("/api/orders/{id}/summary")
public CompletableFuture<OrderSummary> summary(@PathVariable Long id) {
    var pay  = CompletableFuture.supplyAsync(() -> paymentApi.get(id), ioExecutor);
    var ship = CompletableFuture.supplyAsync(() -> shippingApi.get(id), ioExecutor);
    var inv  = CompletableFuture.supplyAsync(() -> invoiceApi.get(id), ioExecutor);
    return CompletableFuture.allOf(pay, ship, inv)
            .thenApply(v -> new OrderSummary(pay.join(), ship.join(), inv.join()))
            .orTimeout(3, TimeUnit.SECONDS);
}

SseEmitter: đẩy trạng thái order.

@Component
public class OrderStatusBroadcaster {
    private final Map<Long, Set<SseEmitter>> subs = new ConcurrentHashMap<>();

    public SseEmitter subscribe(Long orderId) {
        SseEmitter emitter = new SseEmitter(Duration.ofMinutes(30).toMillis());
        subs.computeIfAbsent(orderId, k -> ConcurrentHashMap.newKeySet()).add(emitter);
        Runnable remove = () -> subs.getOrDefault(orderId, Set.of()).remove(emitter);
        emitter.onCompletion(remove);
        emitter.onTimeout(remove);
        emitter.onError(ex -> remove.run());
        return emitter;
    }

    @EventListener                                     // OrderService publish khi đổi trạng thái
    public void on(OrderStatusChanged e) {
        for (SseEmitter em : subs.getOrDefault(e.orderId(), Set.of())) {
            try {
                em.send(SseEmitter.event().id(e.version().toString()).name("status").data(e));
            } catch (IOException ex) {
                em.completeWithError(ex);               // client đã đóng tab
            }
        }
    }
}

@GetMapping(path = "/api/orders/{id}/status/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(@PathVariable Long id) {
    return broadcaster.subscribe(id);
}

Bước 4: Test bằng curl

curl -N localhost:8080/api/orders/42/status/stream
# id:7
# event:status
# data:{"orderId":42,"status":"PACKED"}
#
# id:8
# event:status
# data:{"orderId":42,"status":"SHIPPED"}

curl localhost:8080/api/orders/42/payment       # treo tới khi webhook tới, hoặc 10s → "PENDING"

Phía browser: new EventSource('/api/orders/42/status/stream'), tự reconnect và gửi Last-Event-ID.

5. Cách hoạt động

Tomcat thread #1                        Thread khác (webhook / event / executor)
─────────────────                       ─────────────────────────────────────
DispatcherServlet
 → controller trả DeferredResult
 → request.startAsync()
 → thoát filter chain, thread #1 về pool
          (connection vẫn mở, không thread nào bị giữ)
                                         result.setResult(PAID)
                                         → asyncContext.dispatch()
Tomcat thread #2
 → DispatcherServlet lần 2 (DispatcherType.ASYNC)
 → ghi JSON, đóng response

Lưu ý: filter/interceptor chạy hai lần (REQUEST và ASYNC dispatch); SecurityContext và ThreadLocal không tự sang thread khác, trừ khi được propagate.

6. So sánh

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

• Bật virtual threads (spring.threads.virtual.enabled=true, Java 21): request blocking chỉ giữ một virtual thread rẻ, nên Callable/CompletableFuture chỉ để giải phóng thread Tomcat thường không còn cần. Viết code blocking đơn giản.
• Virtual threads không thay được DeferredResult/SSE theo nghĩa nghiệp vụ: chờ sự kiện ngoài và đẩy nhiều lần vẫn cần chúng (dù giữ một virtual thread cũng rẻ).
• SSE đủ cho thông báo một chiều; cần hai chiều mới lên WebSocket.
• Nhiều instance: emitter nằm trong RAM của từng pod, sự kiện phải phát qua Redis pub/sub hoặc Kafka để pod nào giữ kết nối cũng nhận được.
• Đừng dùng async để "nhanh hơn": nó tăng sức chứa, không giảm latency của một request.

Đúng triệu chứng kinh điển, có hai nguyên nhân:

• Buffering: nginx gom response rồi mới gửi. Tắt cho SSE: proxy_buffering off; hoặc header X-Accel-Buffering: no từ app.
• Idle timeout: proxy_read_timeout mặc định 60s, load balancer cũng có idle timeout. Không có sự kiện trong 60s là bị cắt.

// heartbeat mỗi 20s để giữ kết nối
@Scheduled(fixedRate = 20_000)
void heartbeat() {
    subs.values().forEach(set -> set.forEach(em -> {
        try { em.send(SseEmitter.event().comment("ping")); }
        catch (IOException e) { em.completeWithError(e); }
    }));
}

Và vì kết nối vẫn có thể đứt, client dùng Last-Event-ID để server gửi lại các sự kiện bị lỡ (hoặc ít nhất gửi trạng thái hiện tại khi subscribe).

Bạn hiểu gần đúng, chỉnh một điểm: bỏ những chỗ dùng async chỉ để không chặn thread Tomcat.

• Callable bọc một lời gọi DB/HTTP chậm: bỏ được, viết blocking.
• Gọi song song 3 service: vẫn cần song song, nhưng có thể dùng Executors.newVirtualThreadPerTaskExecutor() hoặc structured concurrency thay vì chuỗi future phức tạp.
• Chờ webhook (DeferredResult) hoặc SSE: vẫn giữ, vì kết quả đến từ thread khác.

Lưu ý pinning: synchronized quanh I/O trên JDK 21–23 có thể ghim virtual thread vào carrier (JDK 24 đã sửa phần lớn). Đo trước khi tin.

Q1. Controller trả DeferredResult thì chuyện gì xảy ra?

🇻🇳 Tiếng Việt:

Spring gọi startAsync của Servlet, trả thread Tomcat về pool nhưng giữ connection mở. Khi code khác gọi setResult, Spring dispatch lại request qua DispatcherServlet trên một thread container để ghi response. Nếu quá timeout thì trả giá trị timeout hoặc lỗi 503. Nhờ vậy hàng nghìn request đang chờ không chiếm thread.

🇬🇧 English:

Spring calls the Servlet startAsync, releases the Tomcat thread but keeps the connection open. When other code calls setResult, Spring dispatches the request again through the DispatcherServlet on a container thread to write the response. On timeout it returns the timeout value or a 503. This way thousands of waiting requests hold no threads.

Q2. Callable khác DeferredResult thế nào?

🇻🇳 Tiếng Việt:

Callable được Spring chạy trên AsyncTaskExecutor, nên vẫn chiếm một thread của executor trong lúc chạy; nó chỉ chuyển việc khỏi pool Tomcat. DeferredResult do ứng dụng tự hoàn thành từ bất kỳ thread nào, như listener message hay webhook, nên không chiếm thread khi chờ.

🇬🇧 English:

Spring runs a Callable on an AsyncTaskExecutor, so it still occupies an executor thread while running; it only moves the work off the Tomcat pool. A DeferredResult is completed by the application from any thread, such as a message listener or a webhook, so no thread is held while waiting.

Q3. Có virtual threads rồi còn cần async MVC không?

🇻🇳 Tiếng Việt:

Phần lớn thì không: với virtual threads, code blocking chỉ giữ một virtual thread rẻ, nên không cần Callable hay CompletableFuture chỉ để giải phóng thread. Nhưng khi kết quả đến từ sự kiện bên ngoài hoặc cần đẩy nhiều sự kiện như SSE, DeferredResult và SseEmitter vẫn là mô hình phù hợp.

🇬🇧 English:

Mostly no: with virtual threads, blocking code only holds a cheap virtual thread, so Callable or CompletableFuture are not needed just to free threads. But when the result comes from an external event or we push many events, like SSE, DeferredResult and SseEmitter are still the right model.

Cách nhớ ngắn gọn

• Async MVC = trả thread, giữ connection, dispatch lại khi có kết quả.
• Callable = executor của Spring; DeferredResult = bạn tự set.
• SSE = nhiều sự kiện một chiều; nhớ heartbeat, tắt buffering, Last-Event-ID.
• Virtual threads xóa lý do "giải phóng thread", không xóa lý do "chờ sự kiện".