Bắt đầu nhanh (5 phút)
Đó là 2:00 sáng ngày Black Friday. Service quản lý sản phẩm của bạn đang xử lý mượt mà ở mức 15.000 requests mỗi giây cho một chiếc màn hình cao cấp đang giảm giá sốc. Redis lưu trữ thông tin tồn kho với TTL 5 phút. Đúng khoảnh khắc key đó hết hạn, 2.500 requests dội vào API của bạn chỉ trong vòng 50 mili-giây. Tất cả đều miss Redis và đâm thẳng xuống PostgreSQL với cùng một câu truy vấn giống hệt nhau: SELECT * FROM products WHERE id = 9812.
CPU của Postgres chạm đỉnh 100%. Latency tăng vọt từ 3ms lên 8.500ms. Chỉ trong vòng 4 giây, toàn bộ 128 kết nối trong connection pool đều cạn kiệt, kéo theo hàng loạt lỗi 504 Gateway Timeout trên toàn cụm hệ thống. Đây chính là hiện tượng thundering herd kinh điển trong thực tế.
Go cung cấp một giải pháp gọn gàng và trực tiếp ngay trong thư viện mở rộng chính thức: golang.org/x/sync/singleflight. Package này giúp khử trùng lặp (deduplicate) các lời gọi hàm đang được thực thi đồng thời. Khi 500 requests cùng yêu cầu một tài nguyên tại một thời điểm, chỉ có duy nhất một query thực sự chạy xuống database. 499 callers còn lại sẽ chỉ việc chờ kết quả đó và dùng chung.
Cài đặt package vào module của bạn:
go get golang.org/x/sync/singleflightDưới đây là một ví dụ hoàn chỉnh có thể chạy được ngay. 5 goroutine cùng truy vấn dữ liệu của một người dùng đồng thời, nhưng thao tác đọc từ database chỉ thực thi đúng một lần duy nhất:
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
"golang.org/x/sync/singleflight"
)
var dbQueryCount int32
func fetchUserDataFromDB(userID string) (string, error) {
atomic.AddInt32(&dbQueryCount, 1)
time.Sleep(150 * time.Millisecond) // Giả lập độ trễ I/O của database
return fmt.Sprintf("Payload thông tin hồ sơ cho %s", userID), nil
}
func main() {
var group singleflight.Group
var wg sync.WaitGroup
userID := "user-4021"
for i := 1; i <= 5; i++ {
wg.Add(1)
go func(workerID int) {
defer wg.Done()
v, err, shared := group.Do(userID, func() (interface{}, error) {
return fetchUserDataFromDB(userID)
})
if err != nil {
fmt.Printf("Worker %d gặp lỗi: %v\n", workerID, err)
return
}
fmt.Printf("Worker %d nhận dữ liệu: %s (shared = %t)\n", workerID, v.(string), shared)
}(i)
}
wg.Wait()
fmt.Printf("\nTổng số query đã thực thi trên DB: %d\n", atomic.LoadInt32(&dbQueryCount))
}Chạy đoạn mã trên, dù có 5 lời gọi cạnh tranh nhau, hàm fetchUserDataFromDB chỉ chạy đúng 1 lần. Caller đầu tiên sẽ gánh vác tác vụ nặng, trong khi 4 caller còn lại nhận được shared = true cùng chính xác payload dữ liệu đó.
Cơ chế hoạt động bên dưới
Toàn bộ package được xây dựng dựa trên một state machine rất tinh gọn. Bên trong singleflight.Group, bạn sẽ thấy một mutex tiêu chuẩn và một hash map nội bộ:
type Group struct {
mu sync.Mutex // bảo vệ map m
m map[string]*call // được khởi tạo lười (lazily initialized)
}Mỗi query đang hoạt động được theo dõi bởi một struct call chứa các primitive đồng bộ hóa, bộ đếm trùng lặp và payload trả về:
type call struct {
wg sync.WaitGroup
val interface{}
err error
dups int
chans []chan<- Result
}Dưới đây là các bước diễn ra tuần tự khi goroutine A gọi group.Do("item-9812", fn):
- Singleflight khóa mutex và kiểm tra xem
m["item-9812"]đã tồn tại hay chưa. - Nếu chưa thấy, nó tạo một
*callmới, tăng 1 vàocall.wg, lưu con trỏ vào map và nhả mutex. - Goroutine A thực thi hàm callback truy xuất dữ liệu
fn().
Nếu goroutine B và C cùng ập đến khi goroutine A vẫn đang đợi database I/O:
- Cả hai routine đều lấy mutex và thấy
m["item-9812"]đã có sẵn trong map. - Chúng tăng biến đếm
call.dups, mở khóa mutex và trực tiếp block tạicall.wg.Wait().
Khi goroutine A hoàn thành, nó ghi kết quả và lỗi vào struct call, gọi call.wg.Done(), khóa mutex lại để xóa key khỏi map rồi mở khóa. Goroutine B và C được đánh thức và lấy kết quả dùng chung. Chỉ với một query database duy nhất đã phục vụ được trọn vẹn ba consumer.
Các pattern thực chiến trên production
Các production service trong thực tế đòi hỏi nhiều thứ hơn là việc block đồng bộ đơn thuần. Bạn cần xử lý timeout, chủ động invalidate cache khi cần, và phòng vệ an toàn cho bộ nhớ.
Xử lý context timeout với DoChan
Hàm group.Do() mặc định sẽ block cho đến khi hàm bên trong chạy xong. Nếu client upstream ngắt kết nối hoặc cài đặt timeout HTTP là 200ms, một query database chậm mất 2 giây sẽ bẫy mọi goroutine đang chờ đợi. Hãy sử dụng group.DoChan() để vừa lắng nghe trên result channel vừa theo dõi request context của bạn:
package cache
import (
"context"
"errors"
"fmt"
"golang.org/x/sync/singleflight"
)
type Service struct {
sf singleflight.Group
}
func (s *Service) GetArticle(ctx context.Context, id string) (string, error) {
resultChan := s.sf.DoChan(id, func() (interface{}, error) {
return fetchFromOrigin(id)
})
select {
case <-ctx.Done():
// Thoát sớm nếu client HTTP hủy request
return "", ctx.Err()
case res := <-resultChan:
if res.Err != nil {
return "", res.Err
}
val, ok := res.Val.(string)
if !ok {
return "", errors.New("kiểu dữ liệu không hợp lệ từ kết quả singleflight")
}
return val, nil
}
}
func fetchFromOrigin(id string) (string, error) {
return fmt.Sprintf("Dữ liệu cho item %s", id), nil
}Lưu ý rằng nếu context của caller đầu tiên hết hạn, tác vụ callback ngầm vẫn sẽ tiếp tục chạy. Singleflight không tự động hủy tác vụ dùng chung này, nhằm đảm bảo các caller khác đang chờ vẫn có thể nhận được dữ liệu.
Bỏ qua các key bị treo hoặc lỗi với group.Forget
Thông thường, singleflight chỉ xóa key sau khi callback đã trả về. Nếu một cổng thanh toán bên thứ ba bị treo trong 30 giây, mọi request gửi đến trong khoảng thời gian đó sẽ bị dồn ứ vào hàng đợi nghẽn mạng này. Gọi s.sf.Forget(id) sẽ yêu cầu group xóa ngay lập tức key đó khỏi map nội bộ.
Worker ban đầu đang chạy dở vẫn tiếp tục ở chế độ nền, nhưng caller tiếp theo ngay sau đó sẽ khởi tạo một query mới thay vì phải xếp hàng chờ đợi tác vụ đang treo. Điều này đặc biệt hữu ích đối với các endpoint hay bị lỗi và cần retry nhanh.
Tránh trả về con trỏ có thể bị biến đổi (mutable pointer)
Lỗi concurrency thường len lỏi qua các con trỏ dùng chung. Do mọi goroutine đang chờ đều nhận được chính xác cùng một interface{}, việc trả về một con trỏ như *UserProfile sẽ cho phép nhiều thread cùng đọc và sửa đổi trên đúng một ô nhớ. Hiện tượng data race chắc chắn sẽ xảy ra.
Luôn ưu tiên trả về dạng value type, clone sâu (deep copy), hoặc chuỗi byte thô đã serialize (chẳng hạn như []byte). Hãy để từng caller tự deserialize ra bản copy cục bộ an toàn của riêng mình.
Những nguyên tắc vận hành xương máu
Qua kinh nghiệm triển khai Singleflight trên nhiều cụm production có thông lượng cao, chúng tôi đúc kết được 4 nguyên tắc sống còn để giữ cho hệ thống luôn ổn định:
- Singleflight làm cổng chắn cho DB; nó không thay thế cache: Luôn kiểm tra Redis hoặc in-memory LRU cache trước tiên. Nếu cache miss, hãy đi qua
singleflight.Groupđể lấy dữ liệu từ PostgreSQL, ghi đè bản ghi mới vào Redis rồi mới trả về. - Đặt namespace cho từng key: Singleflight sử dụng các key kiểu chuỗi (string) đơn thuần. Nếu query người dùng và query hóa đơn cùng tra cứu ID
"4501", chúng sẽ block lẫn nhau và trả về dữ liệu sai lệch. Hãy dùng tiền tố rõ ràng nhưfmt.Sprintf("user:v1:%s", userID). - Phân vùng phạm vi (scope) của group theo domain: Đừng khai báo một
singleflight.Grouptoàn cục duy nhất cho cả microservice. Hãy đặt các instance bên trong từng repository struct tương ứng. Điều này giúp cô lập map lock và loại bỏ tình trạng tranh chấp mutex giữa các domain không liên quan. - Luôn truyền timeout context độc lập cho worker: Nếu query gốc không có deadline, một kết nối database bị deadlock sẽ làm treo toàn bộ subscriber. Luôn thiết lập timeout tường minh cho query (ví dụ:
context.WithTimeout(context.Background(), 3*time.Second)) bên trong hàm truy xuất dữ liệu.
Việc đặt Singleflight phía trước các bảng SQL nặng về đọc và các API bên thứ ba bị giới hạn rate limit sẽ giúp san phẳng các đợt lưu lượng tăng vọt đột ngột một cách đáng tin cậy. Thay vì database bị quá tải bởi hàng loạt thao tác trùng lặp, hệ thống sẽ chỉ xử lý mỗi yêu cầu đúng một lần duy nhất.
