Golang nổi tiếng với mô hình concurrency “share memory by communicating” thay vì “share memory by locking”. Nhưng pattern thông thường như worker pool, pipeline, fan-out/fan-in không tự động xuất hiện khi bạn biết dùng goroutine và channel. Bài này trình bày ba pattern phổ biến nhất với code thực tế, không chỉ snippet học thuật.
Worker Pool pattern
Worker pool là pattern xử lý danh sách job song song với số goroutine cố định. Dùng khi bạn có hàng nghìn task cần xử lý nhưng không muốn spawn hàng nghìn goroutine cùng lúc — vì mỗi goroutine chiếm khoảng 2KB stack, và OS cũng có giới hạn thread.
type Job struct {
ID int
Payload string
}
func worker(id int, jobs <-chan Job, results chan<- JobResult, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
// Xử lý nghiệp vụ, ví dụ gọi API, parse data, resize image
result := process(job)
results <- result
}
}
func main() {
jobs := make(chan Job, 100)
results := make(chan JobResult, 100)
var wg sync.WaitGroup
// Khởi 3 worker
for i := 0; i < 3; i++ {
wg.Add(1)
go worker(i, jobs, results, &wg)
}
// Gửi job
for _, j := range generateJobs() {
jobs <- j
}
close(jobs)
// Chờ worker xong rồi đóng results
wg.Wait()
close(results)
// Đọc kết quả
for r := range results {
fmt.Println("done:", r.ID)
}
}
Điểm mấu chốt: buffer size của channel ảnh hưởng throughput. Buffer nhỏ → worker chờ block, buffer lớn → tốn RAM. Thông thường chọn số worker bằng số CPU logical, buffer bằng 2-4 lần số worker.
Pipeline pattern
Pipeline chia công việc thành các stage nối tiếp, mỗi stage chạy song song với goroutine riêng. Phù hợp xử lý batch: đọc file, transform data, ghi kết quả.
func stage1(in <-chan RawData, out chan<- Parsed) {
for data := range in {
out <- parse(data)
}
close(out)
}
func stage2(in <-chan Parsed, out chan<- Result) {
for p := range in {
out <- enrich(p)
}
close(out)
}
func main() {
raw := make(chan RawData, 50)
parsed := make(chan Parsed, 50)
result := make(chan Result, 50)
go stage1(raw, parsed)
go stage2(parsed, result)
// Feed input
for _, f := range files {
raw <- read(f)
}
close(raw)
// Collect output
for r := range result {
save(r)
}
}
Lợi ích chính: mỗi stage có thể scale độc lập. Nếu parse chậm, tăng worker ở stage1 mà không ảnh hưởng stage2. Backpressure tự nhiên: channel đầy → producer chặn.
Fan-out/Fan-in pattern
Fan-out/fan-in dùng khi bạn cần phân phối một công việc lớn cho nhiều worker, sau đó gộp kết quả. Khác worker pool: input là một channel duy nhất phát sinh liên tục, output cần merge nhiều channel thành một.
func fanOut(tasks <-chan Task) []chan Result {
workers := 4
outs := make([]chan Result, workers)
for i := 0; i < workers; i++ {
outs[i] = make(chan Result)
go func(id int, in <-chan Task, out chan<- Result) {
for t := range in {
out <- doWork(id, t)
}
close(out)
}(i, tasks, outs[i])
}
return outs
}
func fanIn(outs ...<-chan Result) <-chan Result {
var wg sync.WaitGroup
merged := make(chan Result)
wg.Add(len(outs))
for _, ch := range outs {
go func(c <-chan Result) {
defer wg.Done()
for r := range c {
merged <- r
}
}(ch)
}
go func() {
wg.Wait()
close(merged)
}()
return merged
}
Ứng dụng thực tế: crawl hàng trăm URL song song, gọi nhiều API bên thứ ba đồng thời, hoặc chạy inference trên batch ảnh. Số worker nên bằng số CPU logical để tận dụng hiệu năng tối đa mà không gây context switch quá nhiều.
Khi nào dùng pattern nào?
- Worker pool — danh sách job cố định, cần kiểm soát tài nguyên rõ ràng. Ví dụ: resize 10.000 ảnh, gửi 5.000 email.
- Pipeline — dữ liệu chảy qua nhiều stage, mỗi stage biến đổi dữ liệu. Ví dụ: ETL, xử lý log.
- Fan-out/fan-in — một tác vụ lớn cần chia nhỏ và gộp kết quả. Ví dụ: crawl web, batch inference.
Nếu task có I/O bound, số worker có thể lớn gấp 5-10 lần số CPU. Nếu task CPU bound, giữ bằng số CPU logical. Dùng runtime.GOMAXPROCS(runtime.NumCPU()) làm điểm khởi đầu, đo đạc, rồi điều chỉnh.
Lỗi thường gặp
1. Không đóng channel. Producer quên close channel → consumer block mãi ở range. Luôn defer close() sau khi goroutine sinh ra channel xong việc.
2. Deadlock do sync.WaitGroup.Add nằm trong goroutine. Nếu Add gọi sau khi Wait bắt đầu, panic. Luôn gọi Add trước go func.
3. Buffer quá lớn. Buffer 1 triệu không giải quyết bottleneck, chỉ che giấu. Buffer cần vừa đủ để giữ worker bận, không cần hơn.
Tổng kết
Concurrency trong Golang không phải là bắt tất cả chạy song song. N là chọn đúng pattern cho đúng bài toán. Worker pool kiểm soát tài nguyên, pipeline tách biệt concern, fan-out/fan-in tối ưu throughput. Kết hợp cả ba trong một hệ thống lớn hoàn toàn bình thường. Đo đạc latency và throughput trước khi tối ưu — đừng đoán.
Tài liệu tham khảo:
