Go + React: Streaming realtime bằng SSE — backpressure, reconnect và contract type-safe

Có một khoảnh khắc mà hầu hết team fullstack đều đi qua: dashboard cần cập nhật realtime, thế là bạn gõ setInterval(fetch, 3000). Ba tháng sau, hệ thống có 400 user online, bạn nhìn thấy 8,000 request mỗi phút trong log và tự hỏi tại sao con số đó gần như không chứa thông tin gì mới. Lúc đó câu hỏi đúng mới hiện ra. Không phải “dùng WebSocket hay SSE”, mà là chiều dữ liệu của mình thực sự đi hướng nào.

Đường đi thì không ngắn: server Go phát event, client React nhận rồi render, còn đoạn giữa hai đầu là cả một danh sách thứ — và — mà lúc viết demo thì không ai nghĩ tới, đến lúc lên production mới thấy.

Một chiều hay hai chiều: chọn SSE hay WebSocket

WebSocket mạnh ở chỗ song công: hai chiều độc lập cùng lúc. Nghiệp vụ của bạn chỉ có server đẩy xuống, client thỉnh thoảng gọi REST thì sao? Bạn đang trả giá cho một khả năng không dùng: connection stateful thứ hai, cơ chế ping/pong tự viết, tầng auth riêng vì handshake không đi qua middleware HTTP bình thường, và một mớ hạ tầng phải cấu hình lại từ đầu.

Server-Sent Events nhỏ hơn nhiều. Nó chỉ là một HTTP response không bao giờ kết thúc: Content-Type: text/event-stream, vài header nói với proxy đừng đệm, và body là các frame text ngăn nhau bằng dòng trắng. Không protocol mới. Không upgrade handshake. Bù lại nó chạy một chiều thôi — mà nghịch lý là chính giới hạn đó lại làm nó dễ vận hành hơn.

Nguyên tắc chọn gọn lại thế này: client gửi tần suất cao cùng kênh với server đẩy thì WebSocket; client gửi lác đác còn server đẩy liên tục thì SSE; cả hai chiều đều thưa thì polling có điều kiện bằng ETag thường rẻ hơn bất kỳ kênh persistent nào.

Phía Go: một handler, ba thứ dễ quên

Ở tầng thấp nhất, một endpoint SSE không có gì bí ẩn. Nó set header, nó ghi, nó flush, và nó sống chừng nào context của request còn sống.

func (s *Streamer) Handle(w http.ResponseWriter, r *http.Request) {
	rc := http.NewResponseController(w)

	h := w.Header()
	h.Set("Content-Type", "text/event-stream")
	h.Set("Cache-Control", "no-cache, no-transform")
	h.Set("Connection", "keep-alive")
	h.Set("X-Accel-Buffering", "no")

	ctx := r.Context()
	sub := s.hub.Subscribe(ctx, r.PathValue("room"))
	defer s.hub.Unsubscribe(sub)

	// Bắt đầu bằng retry hint: client tự áp dụng khi phải reconnect.
	if _, err := io.WriteString(w, "retry: 3000\n\n"); err != nil {
		return
	}
	if err := rc.Flush(); err != nil {
		return
	}

	heartbeat := time.NewTicker(15 * time.Second)
	defer heartbeat.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-heartbeat.C:
			// Comment frame: giữ connection sống qua proxy, client bỏ qua.
			if _, err := io.WriteString(w, ":hb\n\n"); err != nil {
				return
			}
			if err := rc.Flush(); err != nil {
				return
			}
		case ev, ok := <-sub.C:
			if !ok {
				return
			}
			if _, err := writeEvent(w, ev); err != nil {
				return
			}
			if err := rc.Flush(); err != nil {
				return
			}
		}
	}
}

func writeEvent(w io.Writer, ev Event) error {
	var b bytes.Buffer
	fmt.Fprintf(&b, "id: %d\n", ev.Seq)
	fmt.Fprintf(&b, "event: %s\n", ev.Type)
	for _, line := range strings.Split(ev.Data, "\n") {
		fmt.Fprintf(&b, "data: %s\n", line)
	}
	b.WriteByte('\n')
	_, err := w.Write(b.Bytes())
	return err
}

Ba thứ đó nằm rải rác trong đoạn trên.

Thứ nhất, X-Accel-Buffering: no. nginx mặc định đệm response, chỉ nhả cho client khi buffer đầy hoặc response kết thúc. Response của bạn thì không bao giờ kết thúc, nên client chẳng nhận được gì dù server đã flush sạch sẽ. Cùng vấn đề với một số CDN. Header này là cách nói “đừng đụng vào luồng này”.

Thứ hai, heartbeat. Một TCP connection im lặng đủ lâu sẽ bị cắt ở đâu đó giữa đường: load balancer, NAT, hoặc chính OS. Comment frame — dòng bắt đầu bằng : — mỗi 15 tới 30 giây là cách rẻ nhất để giữ đường sống, và client theo spec phải bỏ qua chúng. Đừng gửi event giả cho việc này. Bạn sẽ phải dạy mọi consumer lọc chúng ra.

Thứ ba, http.NewResponseController. Từ Go 1.20, đây là cách đúng để lấy Flush mà không phụ thuộc vào type-assert http.Flusher, và nó mở đường cho SetWriteDeadline — thứ bạn cần ở phần sau.

Còn một thứ không nằm trong code mà quan trọng ngang: đừng để handler này tự nướng CPU. Poll database trong vòng lặp tức là bạn biến SSE thành một cái máy quay số. Nguồn sự thật phải là hub in-process (channel fan-out) hoặc pub/sub bên ngoài — Redis Pub/Sub, NATS, Postgres LISTEN/NOTIFY — nơi handler chỉ việc ngồi chờ.

Backpressure: chỗ client chậm làm sập server

Đây là phần mọi demo bỏ qua, còn production thì trả bài đủ. Tưởng tượng một room có 2,000 subscriber. Một user mở tab trên laptop cắm mạng khách sạn, hoặc để tab ngủ rồi mở lại sau bốn tiếng. TCP window của họ đầy, w.Write bắt đầu block, và nếu bạn fan-out bằng một channel không giới hạn thì cái block đó lan ngược lên goroutine phát. Một client tệ kéo chậm cả room.

Cách xử lý không nằm ở việc làm channel to hơn. Bounded channel cộng một chính sách rõ ràng khi đầy mới là câu trả lời, và chính sách đó phụ thuộc ngữ nghĩa dữ liệu của bạn.

type Subscription struct {
	C    chan Event
	once sync.Once
	done func()
}

func (s *Subscription) Close() {
	s.once.Do(func() {
		close(s.C)
		s.done()
	})
}

func (h *Hub) Publish(ev Event) {
	h.mu.RLock()
	defer h.mu.RUnlock()

	for sub := range h.subs {
		select {
		case sub.C <- ev:
		default:
			// Client này tụt lại quá xa. Đóng nó, để nó tự reconnect
			// và resume từ offset — rẻ hơn nhiều so với giữ nó sống
			// bằng cách chặn cả hub.
			sub.Close()
			h.metrics.SlowConsumerDropped.Add(1)
		}
	}
}

Consumer quá chậm thì kết thúc nó là hành vi đúng, chứ không tệ. Nhưng chỉ đúng khi client resume được — và id: trong mỗi frame sinh ra vì thế, không phải để trang trí. Nó là con trỏ vị trí. Phát event kèm số thứ tự tăng dần, giữ một ring buffer recent events, và việc ngắt kết nối trở thành non-event: client cầm Last-Event-ID quay lại, server phát bù từ đó.

Cả nhánh đó — client chậm bị ngắt, rồi quay lại xin phát bù — gói gọn trong một sơ đồ:

sequenceDiagram
    participant C as Client React
    participant S as Handler SSE phía Go
    participant H as Hub

    C->>S: GET text/event-stream, nối lại thì kèm Last-Event-ID
    S-->>C: retry: 3000
    loop Mỗi event mới
        H->>S: publish vào channel có giới hạn
        S-->>C: id, event, data rồi Flush
    end
    Note over S: heartbeat 15 giây giữ connection sống qua proxy
    S-->>C: frame comment :hb
    H->>S: channel đầy, client tụt lại quá xa
    S->>C: đóng subscription và ghi metric SlowConsumerDropped
    C->>S: reconnect sau backoff, cầm Last-Event-ID
    S->>H: đọc bù từ ring buffer recent events
    H-->>S: các event còn thiếu
    S-->>C: phát bù rồi Flush

Độ lớn buffer là hàm của tần suất phát và độ trễ chấp nhận được, đừng đoán. Phát 50 event/giây và chấp nhận consumer tụt tối đa 5 giây thì buffer 256 là quá đủ. Ghi lý luận đó vào comment cạnh con số.

Còn một biến thể nữa: nếu dữ liệu là state hiện tại chứ không phải chuỗi sự kiện — giá cổ phiếu, vị trí xe, số người online — thì logic đúng là coalescing, không phải resume. Giữ đúng một giá trị mới nhất cho mỗi key và bỏ các bản trung gian. Nhận diện sai loại dữ liệu thì bạn xây cả một tầng resume vô dụng.

Client React: vì sao EventSource đôi khi không đủ

EventSource là API mặc định và nó có một tài sản quá lớn để bỏ: tự động reconnect kèm header Last-Event-ID. Bạn mất mạng, bạn có lại, browser tự nối lại, và server của bạn — nếu đã có ring buffer — phát bù đúng chỗ. Không một dòng code client nào cho việc đó. Với một dashboard nội bộ, đây gần như luôn là lựa chọn đúng và bạn nên dừng ở đây.

Có đúng hai tình huống nó chặn bạn: cần gửi header tùy chỉnh (Authorization, tenant id) vì EventSource không cho truyền header, hoặc cần POST khi payload filter không nhét nổi vào query string. Cả hai buộc bạn xuống fetch + ReadableStream — và ở đó bạn tự tay viết lại đúng thứ browser vừa mới miễn phí: reconnect, backoff, resume.

type Status = "connecting" | "open" | "closed";

class StreamClient {
  private controller = new AbortController();
  private attempt = 0;
  private lastId: string | null = null;
  private status: Status = "closed";

  constructor(
    private url: string,
    private onEvent: (ev: { id: string; type: string; data: string }) => void,
    private onStatus: (s: Status) => void,
  ) {}

  private setStatus(s: Status) {
    this.status = s;
    this.onStatus(s);
  }

  async start() {
    this.controller = new AbortController();
    this.setStatus("connecting");

    const res = await fetch(this.url, {
      signal: this.controller.signal,
      headers: {
        Accept: "text/event-stream",
        ...(this.lastId ? { "Last-Event-ID": this.lastId } : {}),
      },
    });

    if (!res.ok || !res.body) {
      this.setStatus("closed");
      return this.scheduleRetry();
    }

    this.attempt = 0;
    this.setStatus("open");

    const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
    let buffer = "";

    try {
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        buffer += value;

        // Frame ngăn cách bằng dòng trắng; giữ lại phần dư chưa hoàn chỉnh.
        const frames = buffer.split("\n\n");
        buffer = frames.pop() ?? "";

        for (const frame of frames) {
          const parsed = parseFrame(frame);
          if (!parsed) continue;
          if (parsed.id) this.lastId = parsed.id;
          this.onEvent(parsed);
        }
      }
    } finally {
      reader.releaseLock();
    }

    this.setStatus("closed");
    this.scheduleRetry();
  }

  private scheduleRetry() {
    // Full jitter: tránh cả fleet reconnect cùng một giây sau sự cố.
    const cap = Math.min(30_000, 1000 * 2 ** this.attempt++);
    const delay = cap / 2 + Math.random() * (cap / 2);
    setTimeout(() => void this.start(), delay);
  }

  stop() {
    this.controller.abort();
    this.setStatus("closed");
  }
}

Ba chi tiết trong đoạn này đáng dừng lại.

Buffer phải sống qua các lần read. Một frame SSE có thể bị cắt giữa chừng giữa hai chunk TCP. Parse từng chunk độc lập thì thỉnh thoảng bạn nhận một event rác với data bị cụt — đủ thưa để không ai tái hiện được cho tới lúc production. Giữ buffer, chỉ xử lý tới dòng trắng cuối cùng, mang phần dư sang lần đọc kế. Cùng một lỗi mà người viết parser TCP mắc từ ba mươi năm nay, chỉ khác là giờ nó mặc áo JavaScript.

Full jitter, không phải backoff lũy thừa trần. Server vừa restart và 5,000 client cùng reconnect theo công thức cố định nghĩa là bạn tự DoS chính mình. Ngẫu nhiên hóa khoảng chờ trong biên độ cho phép là một dòng code mua sự bình yên.

Theo dõi status, đừng chỉ theo dõi dữ liệu. Người dùng cần biết khi dữ liệu cũ. Một nhãn “đang kết nối lại” màu vàng trung thực hơn một bảng số trông vẫn đúng nhưng đã đóng băng từ mười phút trước.

Nối SSE vào React mà không đốt re-render

Chỗ dễ mắc nhất sau khi có client là đẩy mọi thứ vào useState trong một component lớn. Với 50 event mỗi giây, bạn ép React re-render cây con đó 50 lần một giây.

Với dữ liệu ngoài React — WebSocket, SSE, một store tự viết, một EventTarget — API chính xác là useSyncExternalStore. Nó cho React biết cách đọc snapshot và cách subscribe, và nó xử lý đúng những ca khó như tearing mà một useEffect cộng useState tự viết sẽ làm sai.

const subscribe = (cb: () => void) => {
  store.listeners.add(cb);
  return () => store.listeners.delete(cb);
};

const getSnapshot = () => store.snapshot;

export function useMetrics(selector: (s: Snapshot) => number) {
  // Trả về số nguyên thay vì object để React so sánh bằng Object.is
  // và chỉ re-render khi giá trị thực sự đổi.
  const value = useSyncExternalStore(subscribe, () => selector(getSnapshot()));
  return value;
}

Chữ đáng bận tâm nhất là selector. Nếu mỗi subscriber nhận cả snapshot, một event chạm một trường cũng làm mọi component re-render. Coi việc chọn store là quyết định kiến trúc chứ không phải chi tiết triển khai: một store nhỏ (Zustand, Jotai, hoặc 60 dòng của bạn) cộng selector granular tốt hơn nhiều so với nhồi toàn bộ stream vào context.

Với danh sách thì ngược lại: đừng append vào mảng dài vô tận. Bounded list — giữ N phần tử mới nhất — bảo vệ cả DOM lẫn bộ nhớ tab. Cần lịch sử thì đó là truy vấn REST, không phải thứ để stream giữ trong RAM.

Contract hai đầu trôi khỏi nhau từ tháng thứ ba

Vấn đề này không cắn ở ngày đầu. Nó cắn ở tháng thứ ba. Server Go có struct Event { Seq uint64; Type string; Data json.RawMessage }, client TypeScript có một interface viết tay, và chúng trôi khỏi nhau khi ai đó đổi tên trường trong Go. TypeScript không báo lỗi vì nó không biết gì về Go. Test không bắt được vì mock cũng viết tay. Chỉ người dùng nhận ra, khi một cột trên dashboard đột nhiên trống.

Cách chữa rẻ nhất là đừng bao giờ viết tay cả hai. Chọn một nguồn sự thật và sinh phần còn lại: mô tả schema bằng IDL — Protobuf, hoặc OpenAPI nếu bạn đã có sẵn — rồi sinh cả struct Go lẫn type TypeScript từ đó. Không muốn thêm build step thì giữ Go làm nguồn sự thật và viết một generator nhỏ in ra file .d.ts. Tool nào không quan trọng bằng quyết định rằng hai đầu được sinh ra, không được gõ.

Hai thứ nên đi kèm bất kể chọn hướng nào:

• Versioning tường minh: nhét một trường v vào envelope và từ chối event có version lạ, biến một lần deploy lệch nhịp thành log rõ ràng thay vì crash khó hiểu.
• Một test đọc schema, không đọc mock: sinh fixture JSON từ chính struct Go trong CI và cho client test parse fixture đó — ai đổi Go mà quên client thì build đỏ ngay trong pull request.

Vài thứ chỉ hiện ra khi lên production

Serverless và SSE cần đọc kỹ điều khoản. Streaming response cần một execution slot sống suốt thời gian kết nối. Trên FaaS, điều đó va vào giới hạn thời gian chạy tối đa của function, và mỗi kết nối mở là một instance bị chiếm. Nó chạy được, nhưng hãy tính chi phí theo số kết nối đồng thời và đặt trần rõ ràng. Với lưu lượng lớn và kết nối dài, một tiến trình Go thường trú sau load balancer hầu như luôn rẻ hơn và bớt bất ngờ hơn.

Mỗi kết nối mở là một goroutine và một file descriptor. 10,000 user online là 10,000 goroutine — Go làm thoải mái — nhưng cũng là 10,000 FD. Biết ulimit của tiến trình trước khi tới con số đó, chứ không phải sau. Đặt metric cho số subscriber và alert khi chạm 80 phần trăm trần đã tính.

Idle timeout ở tầng trên giết connection im lặng. Load balancer nào cũng có idle_timeout, mặc định nhiều nơi là 60 giây; heartbeat 15 giây là để sống chung với nó. Connection chết đều đặn theo chu kỳ tròn trịa gần như chắc chắn là timeout hạ tầng, không phải bug trong code.

Đo thời gian tới client, không chỉ thời gian server xử lý. Một event rời goroutine trong 2ms và tới màn hình sau 400ms là bình thường — và không profile nào của server nhìn thấy khoảng 398ms đó. Muốn biết sự thật thì gắn timestamp vào payload, đo ở client, rồi đẩy số đó ngược về.

Không thêm gì để vận hành

Cách này không nuôi thêm thành phần nào để vận hành: server Go là một handler HTTP, client React là fetch cộng một store. Không broker. Không cổng mới phải mở.

Độ khó nằm hết ở mấy thứ nhỏ — một header chống đệm, một heartbeat, một bounded channel, một buffer sống qua các lần read, một selector đủ hẹp để không re-render cả cây. Toàn đồ cũ. Không có công nghệ mới nào ở đây.

Còn một chỗ mình chưa đo: room nào phát dày hơn 50 event/giây thì con số buffer 256 phải tính lại từ đầu.