Server-Sent Events trong Spring Boot: Redis Pub/Sub và client chậm

Realtime với SseEmitter: xác thực EventSource bằng cookie, fan-out qua nhiều instance, pool ghi có trần biết bỏ gợi ý, và khi nào nên tính tới WebFlux.

· Đăng ngày · Cập nhật · 7 phút đọc · Phiên bản: Spring Boot 3.5, Redis 7 · Về dự án →

Personal OS là app quản lý thời gian chạy trên nhiều thiết bị cùng lúc: sửa lịch trên laptop thì điện thoại cần thấy gần như ngay. Backend viết bằng Spring Boot (Java 21), và phần "realtime" dùng Server-Sent Events (SSE) với SseEmitter của Spring MVC, cộng Redis Pub/Sub khi chạy nhiều instance.

SSE dựng thì nhanh, nhưng chạy ổn định thì cần để ý vài chỗ mà tài liệu ít nhắc tới. Bài này đi qua thiết kế, và chỗ đầu tiên vỡ khi số kết nối tăng. Chỗ đó không nằm ở database, cũng không nằm ở connection pool.

Vì sao SSE mà không phải WebSocket

App chỉ cần một chiều: server báo "có gì đó đổi" cho client. Client gửi dữ liệu lên bằng REST như bình thường. Với nhu cầu này, SSE có vài lợi thế:

  • Chạy trên HTTP thường, đi qua proxy và load balancer dễ dàng.
  • Trình duyệt có sẵn EventSource, tự kết nối lại khi rớt mạng.
  • Phía Spring MVC chỉ cần trả về một SseEmitter, không cần thêm stack mới.

Nguyên tắc: luồng SSE chỉ mang "gợi ý", không mang dữ liệu

Đây là quyết định quan trọng nhất. Mỗi sự kiện SSE chỉ chứa tên bảng vừa thay đổi, ví dụ changed: tasks. Client nhận được thì gọi API đồng bộ (REST) để kéo dữ liệu thật về. Cách API đồng bộ kiểu này giải quyết xung đột được bàn kỹ trong bài đồng bộ cho PWA local-first.

public record SyncEvent(String userId, String table) {}

Nghe có vẻ tốn thêm một request, nhưng đổi lại:

  • Xác thực, phân quyền theo người dùng và logic giải quyết xung đột chỉ nằm ở một chỗ: API đồng bộ. Luồng SSE không cần biết gì về chúng.
  • Mất một sự kiện thì vô hại. Sự kiện chỉ là "hãy pull đi". Client vẫn có nhịp pull đối soát định kỳ, nên bỏ lỡ một gợi ý chỉ làm chậm đi tối đa một chu kỳ, không mất dữ liệu.

Nguyên tắc thứ hai sẽ cứu cả thiết kế ở phần sau.

Xác thực khi EventSource không gửi được header

EventSource của trình duyệt không cho đặt header (constructor EventSource() chỉ nhận tuỳ chọn withCredentials), nên không gửi được Authorization: Bearer .... Có vài cách:

  • Đưa token vào query string: đơn giản nhưng token sẽ nằm trong log của proxy và lịch sử trình duyệt. Nên tránh.
  • Dùng thư viện polyfill thay cho EventSource để gửi header: được, nhưng mất phần tự kết nối lại có sẵn.
  • Cookie HttpOnly: client gọi một endpoint (có xác thực bằng Bearer như mọi API khác) để nhận một cookie HttpOnly chứa token, rồi mở EventSource. Phía server, một filter đọc cookie và kiểm tra token bằng cùng JwtDecoder với phần còn lại của API.

Personal OS dùng cách thứ ba. Filter chỉ chạy khi request chưa được xác thực bằng header, nên luồng Bearer chuẩn vẫn được ưu tiên.

Một đánh đổi cần biết: token chỉ được kiểm tra lúc mở luồng. Luồng mở càng lâu thì một phiên đã bị thu hồi càng giữ kết nối được lâu. Đây là lý do thứ hai khiến nguyên tắc "không mang dữ liệu" quan trọng: luồng chỉ mang tên bảng, còn dữ liệu vẫn phải đi qua REST, nơi mọi request đều được xác thực. Khi đó thời hạn của luồng (ở đây là 60 phút) chỉ còn là chuyện cân bằng giữa số lần kết nối lại và độ trễ thu hồi.

Nhiều instance: Redis Pub/Sub

Mỗi instance chỉ biết những kết nối SSE đang mở tới chính nó. Người dùng sửa dữ liệu qua instance A, nhưng điện thoại của họ lại đang nối vào instance B. Giải pháp là một interface với hai cài đặt:

public interface SyncEventBus {
    void publish(String userId, String table);
}
  • LocalSyncEventBus: chạy một instance, gọi thẳng vào registry trong bộ nhớ.
  • RedisSyncEventBus: publish lên một channel Redis. Mọi instance đều subscribe, và mỗi instance tự gửi tới các kết nối của mình.
Sơ đồ tuần tự với năm bên: laptop, instance A, Redis, instance B và điện thoại. Laptop ghi qua REST vào instance A. A publish user và tên bảng lên Redis. Redis chuyển tới mọi subscriber, trong đó có instance B. B gửi gợi ý SSE changed: tasks tới điện thoại, và điện thoại kéo dữ liệu qua API đồng bộ REST.LaptopAinstanceRedisBinstancePhoneREST: ghipublish(user, tasks)tới mọisubscriberSSE:changed:tasksREST sync:kéo dữ liệu
Luồng SSE chỉ báo bảng nào vừa đổi. Dữ liệu thật vẫn đi qua API đồng bộ REST có xác thực.

Publish lên Redis mà lỗi thì chỉ ghi log cảnh báo. Theo nguyên tắc ở trên, mất một gợi ý thì không sao.

Chỗ vỡ đầu tiên: một client chậm làm chậm tất cả

Có một hiểu lầm phổ biến là "mỗi kết nối SSE giữ một thread". Với Spring MVC thì không hẳn: SseEmitter chạy trên Servlet async, nên thread của request được trả về pool ngay khi controller trả emitter. Kết nối đang mở mà không gửi gì thì không tốn thread nào.

Vấn đề nằm ở lúc gửi. SseEmitter.send() ghi ra socket và block cho tới khi ghi xong. Một client có mạng chậm (bộ đệm TCP đầy) sẽ làm lời gọi này block lâu. Mà trong thiết kế ban đầu, hai đường gửi đều chạy trên thread dùng chung:

  • Heartbeat chạy trên thread của @Scheduled. Mặc định Spring chỉ có một thread cho mọi job @Scheduled (xem cấu hình scheduling mặc định của Spring Boot), dùng chung với các job khác.
  • Gửi gợi ý khi có thay đổi chạy trên thread listener của Redis, cũng chỉ một thread cho cả instance.

Một client chậm vì thế chặn cả vòng gửi. Các client khác trễ heartbeat, proxy tưởng kết nối chết nên cắt, tất cả đồng loạt kết nối lại (reconnect storm), và mỗi lần kết nối lại kéo theo một lượt pull. Đây là giới hạn xuất hiện đầu tiên khi số thiết bị tăng, trước cả database hay connection pool.

Cách sửa: tách việc ghi ra một pool riêng, có trần

Rút gọn từ SyncEventRegistry.java:

private final ThreadPoolExecutor deliverExecutor = new ThreadPoolExecutor(
    4, 4, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(10_000),
    namedDaemonThreadFactory("sse-deliver-"));

private void enqueueSend(String userId, SseEmitter emitter, SseEventBuilder event) {
    try {
        deliverExecutor.execute(() -> send(userId, emitter, event));
    } catch (RejectedExecutionException e) {
        // Hàng đợi đầy: bỏ gợi ý này, client vẫn còn nhịp pull đối soát.
    }
}

Ba lựa chọn đáng chú ý:

  • Pool nhỏ (4 thread). Việc ghi là I/O, không phải CPU, nên vài thread đủ phục vụ nhiều kết nối. Một client chậm chỉ giữ một thread, không chặn thread gọi.
  • Hàng đợi có trần (10.000). Không có trần thì client chậm làm hàng đợi phình ra cho tới khi hết bộ nhớ.
  • Khi đầy thì bỏ, không dùng CallerRunsPolicy. CallerRunsPolicy là lựa chọn "an toàn" quen thuộc, nhưng ở đây "caller" chính là thread heartbeat hoặc thread Redis listener. Để chúng tự chạy việc ghi là dựng lại đúng cái block mình đang muốn tránh. Bỏ gợi ý thì an toàn, nhờ nguyên tắc ở đầu bài.

Ngoài ra, SseEmitter không an toàn khi gửi đồng thời, nên mọi lần gửi tới cùng một emitter phải được tuần tự hoá.

Gộp gợi ý trong 300 ms

Một thao tác lớn phía client (sao chép cả tuần, import dữ liệu) đẩy lên hàng chục batch liên tiếp vào cùng một bảng. Mỗi batch commit lại phát một gợi ý, tức là hàng chục lần ghi SSE cho mỗi thiết bị, và mỗi lần lại kích một lượt pull.

Cách xử lý: các gợi ý cùng (user, table) tới trong vòng 300 ms được gộp thành một lần gửi. Khoảng thời gian này đủ dài để gộp các batch liền nhau, và đủ ngắn để thiết bị khác vẫn thấy thay đổi gần như tức thì. Chi tiết nhỏ: khoá "đang hẹn gửi" được gỡ trước khi gửi, để một gợi ý tới đúng lúc đang gửi sẽ hẹn một lần gửi mới chứ không bị nuốt mất.

Những chi tiết nhỏ khác

  • Heartbeat mỗi 25 giây (một comment SSE, client không thấy) để proxy và load balancer không cắt kết nối "im lặng". Nhiều proxy cắt kết nối rảnh sau khoảng một phút; ví dụ proxy_read_timeout mặc định của Nginx là 60 giây.
  • Sự kiện connected được gửi đồng bộ ngay khi mở luồng, để client biết kết nối đã sẵn sàng. Đây là lần ghi duy nhất không đi qua pool, vì nó chạy trên thread của chính request, cho một emitter vừa tạo.
  • Dọn dẹp emitter ở cả ba callback: onCompletion, onTimeout, onError.

Khi nào cần WebFlux

Với vài trăm thiết bị, thiết kế trên dư sức. Khi số kết nối đồng thời lên hàng nghìn, các giới hạn khác bắt đầu lộ ra: số kết nối tối đa của Tomcat, file descriptor, và chi phí của mô hình "ghi block trên thread". Lúc đó có hai hướng đáng cân nhắc:

  • WebFlux với Flux<ServerSentEvent>: mô hình event-loop, ghi không block, có backpressure.
  • Virtual threads của Java 21 (spring.threads.virtual.enabled=true từ Spring Boot 3.2): giữ nguyên mô hình code blocking nhưng thread trở nên rất rẻ.

Nhờ interface SyncEventBus và việc luồng SSE chỉ mang tên bảng, chuyển sang hướng nào cũng không phải đổi frontend: giao thức text/event-stream và endpoint giữ nguyên.

Thay vì đoán trước khi nào cần chuyển, hãy đo. Các metric đáng theo dõi: số thread JVM đang sống, số kết nối hiện tại của Tomcat, kích thước hàng đợi của pool gửi, và số gợi ý bị bỏ.

Tóm tắt

  • SSE hợp khi server chỉ cần báo cho client rằng có gì đó đã đổi: nó chạy trên HTTP thường, và EventSource tự kết nối lại.
  • Gửi gợi ý, không gửi dữ liệu. Luồng SSE chỉ mang tên bảng, client kéo dữ liệu qua API đồng bộ có xác thực, nên mất một gợi ý chỉ chậm tối đa một chu kỳ đối soát.
  • EventSource không gửi được header. Cookie HttpOnly kiểm tra bằng cùng JwtDecoder giữ token khỏi URL, nhưng token chỉ được kiểm tra lúc mở luồng.
  • Khi chạy nhiều instance, một interface SyncEventBus với cài đặt Redis Pub/Sub đưa mỗi gợi ý tới đúng instance đang giữ kết nối.
  • Chỗ vỡ đầu tiên là send() block trên thread dùng chung: một client chậm làm trễ heartbeat của mọi người. Tách việc ghi ra một pool nhỏ có trần, đầy thì bỏ gợi ý.
  • Gộp gợi ý theo người dùng và bảng trong 300 ms, và gửi heartbeat mỗi 25 giây để proxy không cắt kết nối rảnh.
  • Đo số thread đang sống, số kết nối Tomcat, hàng đợi gửi và số gợi ý bị bỏ trước khi chuyển sang WebFlux hay virtual threads.

Nếu bạn cũng đang làm realtime với Spring và có kinh nghiệm khác, mình rất muốn được trao đổi. Email của mình ở trang chủ.