Fan-in Fan-out trong Go — Chia việc cho goroutine, gom kết quả lại

Mình nhớ hồi làm cái job xử lý lịch sử: một lần phải đọc lại hơn 2 triệu bản ghi từ database cũ, chuẩn hoá dữ liệu rồi ghi qua hệ thống mới. Chạy tuần tự thì ước tính hết... 11 tiếng. Viết xong chương trình fan-in fan-out bằng Go, chạy còn có 40 phút. Hôm nay mình kể lại pattern này — thứ mà hầu như ngày nào mình cũng dùng khi làm backend bằng Go.

Fan-out: chia việc ra cho thiên hạ

Ý tưởng fan-out đơn giản: có một nguồn việc (một channel), mình mở nhiều goroutine cùng ngồi đó, ai rảnh thì bốc một việc ra làm. Thay vì một người làm 2 triệu việc, mình thuê 20 người làm song song.

jobs := make(chan int, 1000)

// Fan-out: 10 worker cùng nhận việc từ 1 channel
var wg sync.WaitGroup
for w := 0; w < 10; w++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        for job := range jobs {
            process(job) // mỗi worker xử lý tuần tự việc của mình
        }
    }(w)
}

// Đẩy việc vào, đóng channel để báo "hết việc"
for i := 0; i < 2_000_000; i++ {
    jobs <- i
}
close(jobs)
wg.Wait()

Chỗ tinh tế nằm ở range jobs — khi channel đóng, vòng lặp tự thoát, worker tự về. Không cần truyền cờ "nghỉ việc" nào cả. Còn close(jobs) sau khi đẩy hết việc là quy ước bất thành văn: chỉ sender được đóng channel, receiver chỉ được đọc.

Fan-in: gom kết quả về một mối

Fan-out mà không gom kết quả thì làm xong cũng như không. Fan-in là ngược lại: nhiều goroutine đẩy kết quả vào chung một channel, một goroutine khác ngồi gom lại.

results := make(chan string, 1000)

// Fan-in: mỗi worker đẩy kết quả vào chung 1 channel
for _, job := range jobs {
    wg.Add(1)
    go func(j Job) {
        defer wg.Done()
        results <- transform(j)
    }(job)
}

// Đóng results khi tất cả worker xong — chỗ này dễ quên!
go func() {
    wg.Wait()
    close(results)
}()

// Gom kết quả
for r := range results {
    writeToDB(r)
}

Cái bẫy kinh điển của fan-in là close(results) đặt sai chỗ. Nếu đóng sớm khi worker còn chạy → panic send on closed channel. Nếu quên không đóng → deadlock vì range results không bao giờ thoát. Mình từng dính cả hai lỗi này, mỗi lần debug mất nửa buổi chiều.

Kinh nghiệm thực chiến

Ngoài pattern cơ bản, có mấy điểm mình đúc kết được:

Giới hạn concurrency. Mở 2 triệu goroutine cùng lúc là chuyện tự sát — Go chạy được nhưng RAM sẽ bốc hơi. Cái worker pool ở trên chính là cách giới hạn: chỉ 10 worker, mỗi worker làm tuần tự, nên tối đa 10 việc chạy cùng lúc. Muốn nhanh hơn thì tăng worker, không phải mở thêm goroutine vô tội vạ.

Dùng errgroup khi cần lỗi. Pattern tay không có một khuyết điểm: worker gặp lỗi thì chỉ biết log, không báo được cho người gọi. Khi cần fail-fast, mình chuyển sang golang.org/x/sync/errgroup:

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10) // giới hạn 10 goroutine chạy song song

for _, job := range jobs {
    job := job
    g.Go(func() error {
        return process(ctx, job) // lỗi đầu tiên sẽ cancel ctx
    })
}
if err := g.Wait(); err != nil {
    log.Fatal("job thất bại: ", err)
}

SetLimit thay luôn cả worker pool lẫn giới hạn concurrency — gọn hơn hẳn pattern thủ công. Còn job := job là để tránh capture biến loop — Go version cũ, hồi đó mình cũng từng dính.

Đừng quên context. Fan-out hàng triệu việc mà không có ctx thì khi muốn dừng (shutdown, timeout) bạn chỉ còn cách kill process. Lúc nào cũng đưa ctx vào process(), kiểm tra ctx.Err() trong vòng lặp để thoát sớm. Kết hợp với errgroup.WithContext thì một lỗi nhỏ cũng kéo cả đội dừng lại đúng lúc.

Kết

Fan-in fan-out nghe sang nhưng thật ra chỉ là: một channel làm nguồn việc, N goroutine làm việc, một channel gom kết quả, đóng channel đúng thời điểm. Pattern này mình dùng hằng ngày — xử lý file, import dữ liệu, gọi API hàng loạt, gửi notification... Còn bạn, hồi nào bạn mới bắt đầu xài goroutine cho việc xử lý song song? Có dính bẫy close(channel) nào chưa?

📋 Phụ lục thuật ngữ

• Fan-out — chia một nguồn tác vụ cho nhiều goroutine xử lý song song
• Fan-in — nhiều goroutine gom kết quả vào chung một channel để xử lý tập trung
• Worker pool — nhóm goroutine cố định cùng nhận việc từ một channel, giới hạn mức độ song song
• errgroup — thư viện Go giúp chạy nhiều goroutine và thu thập lỗi đầu tiên, kèm hỗ trợ context cancel
• Channel — cơ chế giao tiếp giữa các goroutine trong Go, kiểu dữ liệu có hướng truyền nhận

Ảnh: Myburgh Roux — Pexels

Ảnh: Muhammed Ensar — Pexels

Ảnh: Brett Sayles — Pexels