Giao tiếp thời gian thực: Lựa chọn giao thức phù hợp
Các luồng thông báo trực tiếp, bảng giá khớp lệnh và dashboard giám sát hạ tầng đều có chung một yêu cầu: đẩy dữ liệu tức thì về trình duyệt. Cơ chế HTTP polling truyền thống gặp rất nhiều khó khăn ở trường hợp này. Khi 10.000 client liên tục gửi request đến server mỗi giây, điều đó sẽ tạo ra 10.000 request/giây tiêu tốn tài nguyên phân tích cú pháp HTTP header, chi phí TLS overhead và gây quá tải truy vấn cơ sở dữ liệu.
Các kiến trúc hiện đại thường thu hẹp các giải pháp cập nhật thời gian thực vào ba lựa chọn cốt lõi:
- Short/Long Polling: Client gửi request lặp đi lặp lại hoặc giữ kết nối mở cho đến khi có dữ liệu mới. Cơ chế này hoạt động mà không cần hạ tầng đặc biệt, nhưng mỗi lần poll lại gây lãng phí băng thông cho header (thường từ 500B đến 1KB mỗi request) và chiếm dụng các connection slot.
- WebSocket: Nâng cấp kết nối HTTP tiêu chuẩn thành kết nối TCP hai chiều (bidirectional) bền vững. Đây là tiêu chuẩn vàng cho các tác vụ full-duplex như game nhiều người chơi hay bảng vẽ cộng tác thời gian thực. Tuy nhiên, nó đòi hỏi logic tự kết nối lại tùy chỉnh, cơ chế heartbeat và cấu hình proxy riêng biệt.
- Server-Sent Events (SSE): Một tiêu chuẩn nhẹ được xây dựng trực tiếp trên nền tảng HTTP. Server duy trì một response mở liên tục và stream văn bản UTF-8 thuần túy xuống client. Trình duyệt tự động xử lý việc kết nối lại, các proxy doanh nghiệp dễ dàng định tuyến và tận dụng sẵn tính năng multiplexing của HTTP/2 mà không tốn thêm chi phí.
Khi nào nên chọn Server-Sent Events
SSE là lựa chọn lý tưởng cho các luồng dữ liệu một chiều (unidirectional), nơi client chỉ cần nhận dữ liệu mà không cần gửi ngược lại trên cùng một socket.
Ưu điểm nổi bật
- Tương thích mạng hoàn hảo: SSE truyền dữ liệu qua các cổng tiêu chuẩn 80 và 443. Load balancer, API gateway và tường lửa doanh nghiệp xử lý SSE như lưu lượng HTTP thông thường mà không cần handshake nâng cấp WebSocket đặc biệt.
- Khả năng tự phục hồi tích hợp sẵn: API native
EventSourcecủa trình duyệt tự động kết nối lại nếu bị ngắt kết nối. Nó thậm chí còn gửi kèm headerLast-Event-IDđể backend của bạn có thể tiếp tục stream ngay tại vị trí bị gián đoạn. - Dung lượng bộ nhớ cực thấp: Go chỉ cấp phát khoảng 2 KB bộ nhớ stack ban đầu cho mỗi goroutine. Một Go server gọn nhẹ với 4 GB RAM có thể dễ dàng quản lý 50.000 subscriber SSE đồng thời một cách mượt mà.
Đánh đổi & Hạn chế
- Hoàn toàn là một chiều: Dữ liệu chỉ truyền theo một chiều: từ server đến client. Bất kỳ thông điệp nào gửi ngược từ client lên đều cần một HTTP request riêng biệt (như POST hoặc PUT).
- Giới hạn kết nối trình duyệt trên HTTP/1.1: Các trình duyệt áp đặt giới hạn tối đa 6 kết nối HTTP/1.1 mở đồng thời trên mỗi domain. Bạn nên triển khai SSE qua HTTP/2, nơi tính năng multiplexing cho phép hàng trăm stream chạy trên cùng một kết nối TCP duy nhất.
Kiến trúc: Broker Pattern an toàn đa luồng (Thread-Safe) trong Go
Khi hàng ngàn client cùng subscribe đồng thời, việc kiểm soát đồng thời (concurrency control) trở thành thách thức lớn nhất. Những cách triển khai sơ sài thường dẫn đến race condition khi một client ngắt kết nối đúng lúc server đang broadcast dữ liệu.
Việc ghi trực tiếp vào một unbuffered channel hoặc channel đã đóng trong Go sẽ gây ra panic và làm treo (block) các goroutine. Tương tự, thao tác đọc và ghi vào một map client thô từ nhiều goroutine khác nhau sẽ khiến chương trình bị crash ngay lập tức trong runtime.
Mô hình Broker tập trung giải quyết triệt để vấn đề này. Một vòng lặp chuyên trách duy nhất sẽ quản lý registry của client và xử lý tuần tự các sự kiện đăng ký (subscription), hủy đăng ký (unsubscription) và phát sóng (broadcast) thông qua các channel riêng biệt.
Hướng dẫn triển khai từng bước
Dưới đây là cách xây dựng một SSE broker ổn định và chịu lỗi tốt chỉ bằng thư viện chuẩn (standard library) của Go.
Bước 1: Triển khai Broker Thread-Safe
Đầu tiên, định nghĩa state của broker và event loop để quản lý việc đăng ký client cũng như fan-out message một cách an toàn.
package main
import (
"fmt"
"log"
"net/http"
"time"
)
type Client chan string
type Broker struct {
clients map[Client]bool
newClients chan Client
closingClients chan Client
notifier chan string
}
func NewBroker() *Broker {
return &Broker{
clients: make(map[Client]bool),
newClients: make(chan Client),
closingClients: make(chan Client),
notifier: make(chan string),
}
}
func (b *Broker) Start() {
for {
select {
case s := <-b.newClients:
b.clients[s] = true
log.Printf("Client đã kết nối. Số client đang hoạt động: %d", len(b.clients))
case s := <-b.closingClients:
if _, ok := b.clients[s]; ok {
delete(b.clients, s)
close(s)
log.Printf("Client đã ngắt kết nối. Số client đang hoạt động: %d", len(b.clients))
}
case event := <-b.notifier:
for s := range b.clients {
select {
case s <- event:
default:
// Buffer đầy hoặc client xử lý chậm; tránh block toàn bộ broker
log.Println("Bỏ qua client chậm")
}
}
}
}
}Bước 2: Tạo HTTP Handler cho SSE
Tiếp theo, viết HTTP handler. Handler này kiểm tra hỗ trợ flusher, thiết lập các streaming header, đăng ký channel của client và flush dữ liệu ngay khi vừa nhận được.
func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "Không hỗ trợ streaming!", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("Access-Control-Allow-Origin", "*")
messageChan := make(Client, 10)
b.newClients <- messageChan
defer func() {
b.closingClients <- messageChan
}()
ctx := r.Context()
for {
select {
case <-ctx.Done():
return
case msg, ok := <-messageChan:
if !ok {
return
}
// Định dạng chuẩn SSE: data: <payload>\n\n
fmt.Fprintf(w, "data: %s\n\n", msg)
flusher.Flush()
}
}
}Bước 3: Phát dữ liệu đo lường thời gian thực (Live Telemetry)
Khởi chạy broker và gắn thêm một worker chạy ngầm định kỳ gửi thông số CPU metric mỗi hai giây một lần.
func main() {
broker := NewBroker()
go broker.Start()
// Giả lập sự kiện dữ liệu thời gian thực
go func() {
for {
time.Sleep(2 * time.Second)
payload := fmt.Sprintf(`{"timestamp": %d, "metric": "cpu_load", "value": %.2f}`,
time.Now().Unix(),
float64(time.Now().UnixNano()%100)/10.0)
broker.notifier <- payload
}
}()
http.Handle("/events", broker)
log.Println("SSE server đã khởi động trên cổng :8080. Endpoint: /events")
if err := http.ListenAndServe(":8080", nil); err != nil {
log.Fatalf("Server gặp lỗi: %v", err)
}
}Bước 4: Kiểm thử luồng dữ liệu (Stream)
Kiểm tra luồng event thô từ terminal của bạn bằng curl với cờ -N để tắt cơ chế buffer:
curl -N -H "Accept: text/event-stream" http://localhost:8080/eventsViệc nhận và xử lý stream trên trình duyệt chỉ mất vài dòng JavaScript tiêu chuẩn:
const eventSource = new EventSource('http://localhost:8080/events');
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log('Đã nhận metric thời gian thực:', data);
};
eventSource.onerror = (err) => {
console.error('Lỗi kết nối SSE:', err);
};
