SSE, WebSocket, pushSSE, WebSockets, push
Order Service phát sự kiện lên Kafka. Notification Service lưu thông báo vào Postgres rồi publish vào kênh Redis user:{id}; node đang giữ luồng SSE của user đó ghi sự kiện xuống trình duyệt. Bật “Tab kết nối lại” để xem phát lại theo Last-Event-ID, hoặc “User offline” để xem push.The Order Service emits events to Kafka. The Notification Service stores the notification in Postgres, then publishes to the Redis channel user:{id}; the node holding that user's SSE stream writes the event to the browser. Turn on “Tab reconnects” to see replay by Last-Event-ID, or “User offline” to see the mobile push.
1Đơn #5521 chuyển sang SHIPPED. Order Service phát sự kiện, key là user_id để giữ thứ tự theo người dùng.Order #5521 moves to SHIPPED. The Order Service emits an event keyed by user_id to keep per-user order.
Luồng nằm ở một trong nhiều stream node. Node SUBSCRIBE user:{id} khi luồng mở và UNSUBSCRIBE khi đóng; Notification Service chỉ cần PUBLISH. Trên Redis Cluster 7 dùng sharded pub/sub (SSUBSCRIBE/SPUBLISH) để không broadcast qua mọi shard.The stream lives on one of many stream nodes. A node SUBSCRIBEs to user:{id} when a stream opens and UNSUBSCRIBEs when it closes; the Notification Service only PUBLISHes. On Redis Cluster 7 use sharded pub/sub (SSUBSCRIBE/SPUBLISH) so messages are not broadcast across every shard.
Redis Pub/Sub không lưu tin. Lưu thông báo vào DB trước khi publish, gửi id tăng dần trong trường id của SSE, và khi kết nối lại thì đọc id > Last-Event-ID. Subscribe trước rồi mới phát lại, bỏ qua id đã gửi.Redis Pub/Sub keeps nothing. Store the notification before publishing, send an increasing id in the SSE id field, and on reconnect read id > Last-Event-ID. Subscribe first, then replay, and skip ids already sent.
ALB mặc định idle timeout 60 giây, nginx proxy_read_timeout 60 giây. Gửi comment “: ping” mỗi 15 giây, tăng timeout, tắt buffering (X-Accel-Buffering: no) và đặt retry: 3000 để client tự nối lại.An ALB defaults to a 60-second idle timeout and nginx to proxy_read_timeout 60s. Send a “: ping” comment every 15 seconds, raise the timeouts, disable buffering (X-Accel-Buffering: no) and set retry: 3000 so clients reconnect on their own.
EventSource gốc không gửi được header Authorization. Dùng cookie HttpOnly cùng site, hoặc polyfill fetch-event-source để gửi Bearer token; không đặt token trong query string vì nó vào access log. Token hết hạn thì đóng luồng, client làm mới token rồi nối lại.Native EventSource cannot send an Authorization header. Use a same-site HttpOnly cookie, or a fetch-event-source polyfill to send a Bearer token; never put the token in the query string, where it lands in access logs. When the token expires close the stream; the client refreshes it and reconnects.
Stream node giữ key online:{id} với TTL 90 giây, gia hạn mỗi heartbeat. Không có key thì đẩy vào hàng push; worker pool gọi FCM/APNs, retry 429/5xx với backoff và xoá token khi nhận UNREGISTERED.Stream nodes keep online:{id} with a 90-second TTL, refreshed on every heartbeat. With no key the notification goes to the push queue; a worker pool calls FCM/APNs, retries 429/5xx with backoff and deletes tokens that come back UNREGISTERED.
Đếm từ cột read_at với index một phần WHERE read_at IS NULL, hoặc giữ bộ đếm trong Redis và cập nhật cùng lúc. Đánh dấu đã đọc cũng publish một sự kiện để các tab khác giảm số.Count from read_at with a partial index WHERE read_at IS NULL, or keep a Redis counter updated alongside. Mark as read also publishes an event so other tabs drop their count.
| BảngTable | Cột chínhKey columns | Ghi chúNotes |
|---|---|---|
notification | id (bigserial), user_id, type, title, body, link, source_event_id, created_at, read_at | unique(user_id, source_event_id); index (user_id, id) cho phát lại; partial index (user_id) WHERE read_at IS NULL cho số chưa đọcunique(user_id, source_event_id); index (user_id, id) for replay; partial index (user_id) WHERE read_at IS NULL for the unread count |
online:{user_id} (Redis) | value = node_id, TTL 90s | Stream node gia hạn mỗi heartbeat 15–30 giây; hết hạn nghĩa là gửi pushRefreshed by the stream node on every 15–30-second heartbeat; expiry means send a push |
user:{user_id} (Redis channel) | payload = { id, type, title, link } | Chỉ để đánh thức luồng đang mở; nguồn sự thật là bảng notificationOnly wakes open streams; the notification table is the source of truth |
device_token | user_id, platform, token, last_seen_at | Một user nhiều thiết bị; xoá token khi FCM trả UNREGISTEREDMany devices per user; delete a token when FCM answers UNREGISTERED |
Endpoint SSE đọc Last-Event-ID, phát lại thông báo đã lỡ, sau đó gửi sự kiện trực tiếp và ping định kỳ để load balancer không cắt kết nối.The SSE endpoint reads Last-Event-ID, replays missed notifications, then streams live events and pings regularly so the load balancer keeps the connection open.
@RestController
@RequiredArgsConstructor
public class NotificationStreamController {
private final NotificationRepository repo;
private final EmitterRegistry emitters; // SUBSCRIBE user:{id} on first emitter
@GetMapping(path = "/notifications/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(@AuthenticationPrincipal Jwt jwt,
@RequestHeader(value = "Last-Event-ID", required = false) Long lastId) {
String userId = jwt.getSubject();
SseEmitter emitter = new SseEmitter(Duration.ofMinutes(30).toMillis());
emitters.add(userId, emitter); // live events from now on
emitter.onCompletion(() -> emitters.remove(userId, emitter));
emitter.onTimeout(emitter::complete);
if (lastId != null) {
repo.findTop100ByUserIdAndIdGreaterThanOrderById(userId, lastId)
.forEach(n -> send(emitter, n)); // replay what was missed
}
return emitter;
}
@Scheduled(fixedRate = 15_000)
public void heartbeat() {
emitters.forEach((userId, e) -> {
try {
e.send(SseEmitter.event().comment("ping"));
emitters.touchOnline(userId, Duration.ofSeconds(90));
} catch (IOException ex) {
emitters.remove(userId, e);
}
});
}
static void send(SseEmitter e, Notification n) {
try {
e.send(SseEmitter.event()
.id(Long.toString(n.getId()))
.name("notification")
.data(n.toDto())
.reconnectTime(3_000));
} catch (IOException ex) {
e.completeWithError(ex);
}
}
}func (s *Server) Stream(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
userID := auth.UserID(ctx) // set by the JWT middleware
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("X-Accel-Buffering", "no") // nginx: do not buffer
sub := s.redis.Subscribe(ctx, "user:"+userID)
defer sub.Close()
if _, err := sub.Receive(ctx); err != nil { // wait for the SUBSCRIBE ack
return
}
live := sub.Channel()
lastID, _ := strconv.ParseInt(r.Header.Get("Last-Event-ID"), 10, 64)
if lastID > 0 { // reconnect: replay what was missed
missed, _ := s.store.After(ctx, userID, lastID, 100)
for _, n := range missed {
writeEvent(w, n)
lastID = n.ID
}
}
flusher.Flush()
ping := time.NewTicker(15 * time.Second)
defer ping.Stop()
for {
select {
case msg := <-live:
n := decode(msg.Payload)
if n.ID <= lastID {
continue // already sent during replay
}
writeEvent(w, n)
lastID = n.ID
case <-ping.C:
fmt.Fprint(w, ": ping\n\n")
s.presence.Touch(ctx, userID, 90*time.Second) // SET online:{u} EX 90
case <-ctx.Done():
return // tab closed or the client went away
}
flusher.Flush()
}
}
func writeEvent(w io.Writer, n Notification) {
fmt.Fprintf(w, "id: %d\nevent: notification\ndata: %s\n\n", n.ID, n.JSON)
}| SSESSE | WebSocketWebSocket | |
|---|---|---|
| Chiều dữ liệuDirection | Server → client; client gọi REST khi cần ghiServer to client; the client uses REST to write | Hai chiều trên một kết nốiBoth ways on one connection |
| Nối lại, phát lạiReconnect and resume | Có sẵn: EventSource tự retry và gửi Last-Event-IDBuilt in: EventSource retries and sends Last-Event-ID | Tự viết: backoff, gửi id cuối, phát lạiWrite it yourself: backoff, send the last id, replay |
| Hạ tầngInfrastructure | HTTP thường, qua mọi proxy; HTTP/1.1 giới hạn 6 kết nối mỗi domain, HTTP/2 thì ghép kênhPlain HTTP through any proxy; HTTP/1.1 allows 6 connections per domain, HTTP/2 multiplexes | Cần Upgrade được LB và proxy hỗ trợ; không có giới hạn 6 kết nốiNeeds Upgrade support in the LB and proxies; no 6-connection limit |
| Chọn khiPick it when | Chỉ đẩy thông báo, cập nhật trạng tháiYou only push notifications and status updates | Đã có chat, cộng tác hoặc cần độ trễ hai chiều thấpYou already have chat or collaboration, or need low-latency both ways |
Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.
SSE trong đa số trường hợp: một chiều, HTTP thường, có sẵn retry và Last-Event-ID. WebSocket khi đã có kênh hai chiều (chat). Long polling là phương án dự phòng cho proxy chặn streaming, đổi lại mỗi sự kiện tốn một request mới và độ trễ cao hơn.SSE in most cases: one-way, plain HTTP, with retry and Last-Event-ID built in. WebSocket when a two-way channel (chat) already exists. Long polling is the fallback for proxies that block streaming, at the cost of a new request per event and higher latency.
Pub/Sub không lưu gì: tab đang nối lại sẽ lỡ sự kiện. DB cho id tăng dần để phát lại, cho danh sách chuông và số chưa đọc. Publish chỉ là lời đánh thức.Pub/Sub stores nothing, so a reconnecting tab would miss the event. The DB gives an increasing id for replay, the bell list and the unread count. The publish is only a wake-up call.
50k client nối lại cùng giây và cùng truy vấn phát lại. Gửi retry ngẫu nhiên 2–10 giây trước khi tắt node, drain từ từ, giới hạn phát lại 100 bản ghi và dùng index (user_id, id).50k clients reconnect in the same second and all run the replay query. Send a randomised retry of 2–10 seconds before shutting the node down, drain gradually, cap replay at 100 rows and use an (user_id, id) index.
Cookie HttpOnly SameSite cùng domain là cách gọn nhất. Nếu bắt buộc Bearer token thì dùng fetch-event-source. Tránh token trong URL; nếu buộc phải dùng thì là ticket ngắn hạn một lần, đổi từ API trước khi mở luồng.A same-domain HttpOnly SameSite cookie is the simplest. If a Bearer token is required, use fetch-event-source. Avoid tokens in the URL; if you must, use a short-lived one-time ticket fetched from the API just before opening the stream.
Phím 1–4 để chọn đáp án, Enter để kiểm tra.Press 1–4 to choose an answer, Enter to check.
Ôn tập ngắt quãng (spaced repetition): thẻ được lặp lại sau 1 → 3 → 7 → 14 → 30 ngày để củng cố trí nhớ dài hạn.Spaced repetition: cards are repeated after 1 → 3 → 7 → 14 → 30 days to build long-term memory.