Go Zero to Hero - Bài 69: Concurrency Pattern: Fan-in và Fan-out

1. Bản chất của Fan-out (Tỏa ra) và Fan-in (Hội tụ)

Đây là mẫu thiết kế (Pattern) sinh ra để giải quyết bài toán tải nặng cục bộ trong một Pipeline:

• Fan-out (Chia để trị): Khi một công đoạn quá chậm, ta tạo ra NHIỀU Goroutines (Workers) cùng đọc chung từ một Channel đầu vào. Dữ liệu từ ống nước sẽ tỏa ra (Fan-out) đều cho các Worker rảnh rỗi xử lý.
• Fan-in (Hợp giang): Các Worker sau khi xử lý xong sẽ nhét kết quả vào các Channel đầu ra khác nhau. Việc của ta là phải viết một Goroutine gom tất cả các ống đầu ra đó lại, hòa chung vào một Channel duy nhất để công đoạn tiếp theo dễ dàng tiêu thụ.

Hai kỹ thuật này gần như luôn phải đi đôi với nhau trong kiến trúc Microservices và xử lý Data Streaming.

2. Kiến trúc giải quyết nút cổ chai

Hãy xem cách chúng ta thiết kế luồng xử lý giao dịch vé tàu. Công đoạn làm giàu dữ liệu (Enrich) tốn rất nhiều thời gian, ta sẽ Fan-out nó ra 3 Workers.

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// CÔNG ĐOẠN 1: TẠO DỮ LIỆU (Nhanh)
func generateJobs(ctx context.Context, jobs []int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, j := range jobs {
            select {
            case out <- j:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// CÔNG ĐOẠN 2: WORKER CHẬM CHẠP (Nút thắt cổ chai)
func enrichWorker(ctx context.Context, workerID int, in <-chan int) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for job := range in {
            // Giả lập gọi DB tốn 100ms
            time.Sleep(100 * time.Millisecond)
            result := fmt.Sprintf("[Worker %d] Đã làm giàu dữ liệu vé #%d", workerID, job)
            
            select {
            case out <- result:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// CÔNG ĐOẠN 3: FAN-IN (Gộp các ống nước lại thành 1)
func merge(ctx context.Context, channels ...<-chan string) <-chan string {
    var wg sync.WaitGroup
    out := make(chan string)

    // Hàm tiện ích để chuyển nước từ ống con vào ống cái
    output := func(c <-chan string) {
        defer wg.Done()
        for val := range c {
            select {
            case out <- val:
            case <-ctx.Done():
                return
            }
        }
    }

    // Lặp qua từng ống con, mỗi ống tung 1 Goroutine để hút nước sang ống cái
    wg.Add(len(channels))
    for _, c := range channels {
        go output(c)
    }

    // Đợi tất cả ống con cạn kiệt thì đóng ống cái
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    jobs := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

    // 1. Khởi động nguồn phát
    inStream := generateJobs(ctx, jobs)

    // 2. FAN-OUT: Tỏa ra 3 Workers để tăng tốc
    var workers []<-chan string
    for i := 1; i <= 3; i++ {
        workers = append(workers, enrichWorker(ctx, i, inStream))
    }

    // 3. FAN-IN: Gộp 3 ống kết quả lại thành 1 ống duy nhất
    mergedStream := merge(ctx, workers...)

    // 4. TIÊU THỤ: Đọc từ ống hợp nhất
    for result := range mergedStream {
        fmt.Println("✅", result)
    }

    fmt.Println("--- TOÀN BỘ DÂY CHUYỀN HOÀN TẤT ---")
}

Nếu chạy đoạn code trên, 10 công việc đáng lẽ tốn 1000ms (1 giây) nếu chạy tuần tự, nay chỉ tốn vỏn vẹn khoảng 300ms - 400ms nhờ sự hợp lực song song của 3 Workers.

3. Chìa khóa sinh tử trong hàm merge (Fan-in)

Nhìn vào hàm merge ở trên, bạn sẽ thấy đây là một kiệt tác của việc kết hợp Goroutine, Channel và sync.WaitGroup. Có 2 cái bẫy bạn cần né khi viết hàm Fan-in:

1. Tuyệt đối không đợi đồng bộ ở luồng chính của merge: Nếu bạn viết wg.Wait() và close(out) ở ngay luồng chính của hàm merge mà không bọc nó trong một go func(), hàm merge sẽ bị khóa chặn (Block) lại. Nó sẽ không thể trả về cái ống out cho hàm main sử dụng, dẫn đến Deadlock toàn bộ chương trình.
2. Mỗi ống con phải có một Goroutine hút dữ liệu độc lập: Nếu bạn cố dùng chung một vòng lặp for để đọc lần lượt từng ống con, bạn lại vô tình biến việc đọc song song thành đọc tuần tự. Phải luôn tuân thủ nguyên tắc: 1 ống con = 1 Goroutine phụ trách đổ vào ống cái.

4. Khi nào thì dùng Pattern này?

Fan-out / Fan-in là vũ khí tối thượng, nhưng không phải lúc nào cũng nên lôi ra dùng.

✅ Chỉ dùng khi:

• Công đoạn đó liên quan đến I/O (gọi Database MySQL/PostgreSQL, gọi API bên thứ 3, đọc file đĩa).
• Khối lượng dữ liệu cực lớn cần chia nhỏ tải.

❌ Không dùng khi:

• Công đoạn đó chỉ tính toán bằng CPU (chẳng hạn cộng trừ nhân chia đơn giản). Việc tạo quá nhiều Goroutines để tính toán những hàm chỉ tốn vài nano-giây sẽ khiến thời gian CPU chuyển đổi ngữ cảnh (Context Switching) còn lâu hơn cả thời gian chạy hàm, làm hệ thống chậm đi chứ không hề nhanh lên.

💡 Tổng kết

Bạn đã nắm vững nghệ thuật điều phối luồng dữ liệu phân tán:

• Fan-out: Phân phát dữ liệu từ 1 ống cái ra nhiều ống con cho các Worker xử lý song song, đập tan nút cổ chai I/O.
• Fan-in: Thu thập kết quả từ nhiều ống con gom lại thành 1 ống cái để dễ dàng báo cáo hoặc chuyển sang công đoạn kế tiếp.
• Công cụ bắt buộc: sync.WaitGroup là trái tim của Fan-in để biết khi nào thì tất cả các dòng chảy con đã cạn kiệt.