Go Zero to Hero - Bài 67: Concurrency Pattern: Worker Pool

1. Worker Pool là gì? Giới hạn sức mạnh để sinh tồn

Worker Pool (Hồ chứa công nhân) là một mô hình nơi bạn tạo ra một số lượng cố định các Goroutines (gọi là Worker), hoạt động liên tục ngầm dưới nền.

Thay vì cứ mỗi một "Công việc" (Job) đến, bạn lại đẻ ra một Goroutine mới, thì bạn sẽ ném tất cả Job đó vào một cái Hàng đợi (Buffered Channel). Các Worker rảnh rỗi sẽ thi nhau thò tay vào cái Hàng đợi đó để rút Job ra làm. Làm xong, nó lại tiếp tục rút Job khác cho đến khi Hàng đợi cạn kiệt.

Lợi ích cốt lõi:

• Throttling (Điều tiết): Kiểm soát chính xác số lượng luồng chạy đồng thời. Nếu PostgreSQL chỉ chịu được tối đa 50 kết nối, bạn set số lượng Worker là 50. Hệ thống sẽ luôn an toàn tuyệt đối.
• Chống rác bộ nhớ: Tránh việc khởi tạo và hủy Goroutine liên tục gây quá tải cho Garbage Collector.

2. Ba trụ cột của một Worker Pool chuẩn mực

Một Worker Pool vững chãi luôn được cấu thành từ 3 thành phần:

1. Jobs Channel: Ống nước chứa các công việc cần làm (Thường là Buffered Channel).
2. Results Channel: Ống nước chứa kết quả trả về từ các Worker.
3. Worker Function: Một hàm chạy vòng lặp vô tận (cho đến khi Jobs Channel đóng), liên tục lấy việc ra làm.

3. Thực chiến: Hệ thống quyết toán hàng triệu lượt vé AFC

Hãy thiết kế một Worker Pool để xử lý một mẻ dữ liệu khổng lồ gồm 10 giao dịch (Job), nhưng chúng ta chỉ thuê đúng 3 Worker để đảm bảo Database không bị nghẽn.

package main

import (
    "fmt"
    "time"
)

// 1. WORKER: Công nhân cày cuốc
// Nhận vào Jobs Channel (chỉ lấy dữ liệu) và Results Channel (chỉ nhét dữ liệu)
func ticketWorker(id int, jobs <-chan int, results chan<- string) {
    // Vòng lặp for range sẽ liên tục rút việc từ ống jobs.
    // Tự động thoát khi ống jobs bị ĐÓNG (close).
    for j := range jobs {
        fmt.Printf("👷 [Worker %d] Đang xử lý giao dịch vé số #%d...\n", id, j)
        
        // Giả lập thời gian ghi xuống Database tốn 1 giây
        time.Sleep(1 * time.Second) 
        
        // Đẩy kết quả vào ống results
        results <- fmt.Sprintf("Vé #%d hoàn tất bởi Worker %d", j, id)
    }
    fmt.Printf("🛑 [Worker %d] Hết việc, xin phép đi ngủ!\n", id)
}

func main() {
    const numJobs = 10
    const numWorkers = 3

    // Khởi tạo 2 đường ống với sức chứa bằng tổng số lượng công việc
    jobs := make(chan int, numJobs)
    results := make(chan string, numJobs)

    fmt.Println("--- BẮT ĐẦU MỞ CỬA NHÀ MÁY ---")

    // 2. KÍCH HOẠT WORKER POOL
    // Tung ra đúng 3 Worker chạy ngầm chờ chực dữ liệu
    for w := 1; w <= numWorkers; w++ {
        go ticketWorker(w, jobs, results)
    }

    // 3. ĐIỀU PHỐI VIÊN (DISPATCHER) GIAO VIỆC
    // Ném 10 công việc vào Hàng đợi
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    
    // ĐÓNG ỐNG JOBS: 
    // Báo hiệu không còn việc mới nữa. Các Worker làm nốt số hàng tồn kho rồi sẽ tự tắt.
    close(jobs)

    // 4. NGƯỜI QUẢN LÝ NGHIỆM THU KẾT QUẢ
    // Rút chính xác 10 kết quả từ ống Results
    for a := 1; a <= numJobs; a++ {
        res := <-results
        fmt.Println("✅ [Nghiệm thu]", res)
    }
    
    fmt.Println("--- CHỐT SỔ, ĐÓNG CỬA NHÀ MÁY ---")
}

Phân tích luồng chạy: Khi bạn chạy đoạn code này, bạn sẽ thấy dù có 10 công việc, nhưng tại mọi thời điểm, chỉ có đúng 3 giao dịch được xử lý cùng lúc. Nút cổ chai (Bottleneck) đã bị bóp nghẹt một cách có chủ đích ở tầng Application, cứu sống tầng Database.

4. Tuyệt kỹ nâng cao: Kết hợp Worker Pool và WaitGroup

Đoạn code trên chạy rất ổn nếu bạn biết trước chính xác số lượng Job (để rút đúng numJobs kết quả). Nhưng trong thực tế, nếu luồng Job là động (đọc từ Kafka/RabbitMQ), bạn phải dùng sync.WaitGroup để chốt chặn và đóng Channel an toàn.

Đây là bộ khung (Boilerplate) "đáng giá ngàn vàng" mà mọi Senior Go Developer đều thuộc lòng:

func advancedWorkerPool() {
    jobs := make(chan string, 100)
    results := make(chan string, 100)
    var wg sync.WaitGroup

    // 1. Tạo Pool 5 Workers
    for w := 1; w <= 5; w++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for job := range jobs {
                // Xử lý job...
                results <- job + " - DONE"
            }
        }(w)
    }

    // 2. Một Goroutine làm nhiệm vụ DỌN DẸP ngầm
    // Nó sẽ chờ 5 Workers làm xong hết, rồi tự động ĐÓNG ống Results
    go func() {
        wg.Wait()
        close(results)
    }()

    // 3. Goroutine đẩy việc (Từ file, DB, Message Broker)
    go func() {
        for i := 0; i < 20; i++ {
            jobs <- fmt.Sprintf("Mã_%d", i)
        }
        close(jobs) // Đẩy xong thì đóng ống jobs
    }()

    // 4. Main luồng hứng kết quả an toàn bằng for range
    for res := range results {
        fmt.Println("Hoàn thành:", res)
    }
}

Nhờ cái "Goroutine Dọn dẹp" kẹp wg.Wait(), toàn bộ hệ thống tự động tắt điện, thu hồi RAM gọn gàng mà không bao giờ bị Deadlock.

📊 Bảng tổng kết Kiến trúc