Worker pool trong Go: giới hạn goroutine, thu kết quả và hủy an toàn

1. Tổng quan

Goroutine trong Go rất rẻ, nên việc gọi go cho mỗi tác vụ là chuyện thường gặp. Khi số tác vụ lên tới hàng chục nghìn và mỗi tác vụ gọi database hoặc HTTP, cách làm này dễ làm cạn connection pool hoặc vượt rate limit của dịch vụ bên ngoài.

Bài viết này hướng dẫn bạn xây dựng worker pool, tức một số lượng cố định goroutine cùng nhận việc từ một channel. Đọc xong, bạn sẽ biết cách viết pool cơ bản, thu kết quả và lỗi, hủy bằng context, rút gọn bằng errgroup.SetLimit, và kiểm thử rằng pool thật sự giới hạn mức song song.

2. Vấn đề: goroutine không giới hạn

Xét đoạn code xử lý danh sách đơn hàng:

for _, o := range orders {
	go chargeOrder(o) // 50.000 order -> 50.000 goroutine cùng gọi payment API
}

Mỗi goroutine chỉ tốn vài KB bộ nhớ, nên vấn đề thường không nằm ở RAM. Vấn đề nằm ở tài nguyên phía sau:

• Connection pool của database chỉ có khoảng 20–50 kết nối.
• Payment gateway giới hạn số request mỗi giây.
• Số file descriptor của tiến trình có giới hạn.

Worker pool giải quyết việc này bằng cách tách hai khái niệm. Số tác vụ có thể rất lớn, còn số tác vụ chạy cùng lúc thì bị giới hạn bởi số worker.

3. Worker pool cơ bản

Mô hình gồm ba phần:

1. Một channel jobs chứa việc cần làm.
2. N goroutine worker, mỗi worker lặp for range trên channel.
3. Một sync.WaitGroup để chờ mọi worker kết thúc.

package main

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

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // giả lập gọi payment API
		fmt.Printf("worker %d xử lý order %d (%.0f)\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("xong sau", time.Since(start).Round(100*time.Millisecond))
}

Đoạn code hoạt động như sau:

• orders <-chan Order là channel chỉ nhận ở phía worker. Compiler sẽ chặn worker nếu nó lỡ gửi vào hoặc đóng channel.
• for o := range orders tự thoát khi channel bị close và đã rút hết phần tử.
• close(orders) là tín hiệu “hết việc”. Nếu bạn quên dòng này, worker chờ mãi và wg.Wait() gây deadlock.
• wg.Add(1) phải gọi trước go worker(...), không gọi bên trong goroutine.

Kết quả chạy (thứ tự dòng có thể khác):

worker 3 xử lý order 1 (10)
worker 1 xử lý order 2 (20)
worker 2 xử lý order 3 (30)
worker 2 xử lý order 4 (40)
...
xong sau 300ms

9 order chia cho 3 worker, mỗi order mất 100ms, nên tổng thời gian khoảng 300ms thay vì 900ms nếu chạy tuần tự.

Lưu ý: Từ Go 1.25, bạn có thể viết wg.Go(func() { worker(i, orders) }). Hàm này tự gọi Add và Done, nên tránh được lỗi quên một trong hai.

4. Thu kết quả và lỗi

Trong thực tế, bạn thường cần biết order nào thành công và order nào lỗi. Ta thêm một channel results cho việc này:

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("số tiền không hợp lệ")
	}
	return o.Amount * 0.02, nil // phí 2%
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for r := range results {
		if r.Err != nil {
			fmt.Printf("order %d lỗi: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("order %d phí %.2f\n", r.OrderID, r.Fee)
	}
}

Đầu ra có dạng:

order 1 phí 2.00
order 3 phí 5.00
order 2 lỗi: số tiền không hợp lệ
order 4 phí 1.60

Có ba điểm cần chú ý:

• Việc gửi job chạy trong goroutine riêng. Nếu main vừa gửi job vừa đọc kết quả trên cùng một goroutine, worker sẽ bị chặn ở results <- trong khi main bị chặn ở jobs <-. Kết quả là deadlock.
• Goroutine thứ ba chờ wg.Wait() rồi mới close(results). Chỉ khi mọi worker xong thì mới chắc chắn không còn ai gửi vào results. Quy tắc chung: chỉ phía gửi mới đóng channel.
• Lỗi được gói trong Result thay vì làm dừng cả pool. Lỗi của một order không nên chặn các order khác.

Thứ tự kết quả không trùng với thứ tự đầu vào. Nếu cần giữ thứ tự, hãy mang theo chỉ số (index) trong job, hoặc dùng cách ở mục 6.

5. Hủy pool bằng context

Khi request HTTP bị hủy hoặc hết timeout, pool phải dừng lại thay vì tiếp tục gọi payment API. Bạn làm việc này bằng cách để worker lắng nghe ctx.Done():

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

Phía producer cũng cần select tương tự:

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

Và tạo context có timeout ở nơi gọi:

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

Lý do cần select ở cả hai chiều gửi: nếu worker chỉ kiểm tra ctx khi nhận job, nó vẫn có thể bị kẹt mãi ở results <- khi phía đọc đã bỏ đi. Đây là nguồn rò rỉ goroutine (goroutine leak) rất phổ biến.

Lưu ý: select chỉ giúp worker thoát giữa các job. Nếu chargeOrder gọi HTTP, bạn cần truyền ctx vào nó (http.NewRequestWithContext) để hủy được cả request đang chạy dở.

6. Rút gọn với errgroup.SetLimit

Viết tay channel, WaitGroup và context dễ sai. Gói errgroup có sẵn SetLimit, cho phép tạo pool có giới hạn chỉ trong vài dòng. Trước hết, thêm dependency:

go get golang.org/x/sync/errgroup

Hàm generic dưới đây xử lý một slice với tối đa limit goroutine, giữ nguyên thứ tự kết quả, và dừng ở lỗi đầu tiên:

package pool

import (
	"context"

	"golang.org/x/sync/errgroup"
)

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

Đoạn code hoạt động như sau:

• g.SetLimit(limit) khiến g.Go bị chặn khi đã có đủ limit goroutine đang chạy. Về hiệu quả, nó tương đương một worker pool.
• errgroup.WithContext trả về ctx sẽ bị hủy ngay khi có một hàm trả lỗi. Các tác vụ khác đang dùng ctx sẽ nhận được tín hiệu dừng.
• results[i] = r an toàn khi chạy đồng thời vì mỗi goroutine ghi vào một chỉ số riêng. Không cần mutex trong trường hợp này.
• Từ Go 1.22, biến vòng lặp i và item được tạo mới ở mỗi vòng, nên closure không còn bị dùng chung biến như các phiên bản cũ.

Cách gọi:

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

Khác biệt so với pool tự viết: errgroup tạo một goroutine mới cho mỗi item thay vì tái sử dụng N worker. Với đa số ứng dụng, chênh lệch này không đáng kể. Nếu bạn cần worker giữ trạng thái riêng, chẳng hạn mỗi worker có một connection, thì hãy dùng pool tự viết ở mục 3–5.

7. Kiểm thử worker pool

Bạn cần kiểm thử hai hành vi chính. Thứ nhất, pool không bao giờ vượt quá limit. Thứ hai, lỗi được trả về đúng. Để đo mức song song, ta dùng atomic.Int32:

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("không mong đợi lỗi: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("chạy đồng thời %d, giới hạn là 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("thẻ bị từ chối")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("muốn errDeclined, nhận %v", err)
	}
}

Chạy test kèm race detector:

go test -race ./...

ok  	example.com/shop/pool	0.061s

Vòng CompareAndSwap cập nhật maxSeen mà không cần khóa. Cờ -race là bắt buộc với code đồng thời, vì nó phát hiện các truy cập dữ liệu chung mà mắt thường dễ bỏ sót.

8. Lỗi hay gặp và khi nào không nên dùng

Các lỗi thường gặp:

• Quên close(jobs): worker không bao giờ thoát và chương trình báo fatal error: all goroutines are asleep - deadlock!.
• Đóng channel ở phía nhận, hoặc đóng hai lần: gây panic: close of closed channel.
• Gọi wg.Add bên trong goroutine: Wait có thể trả về trước khi goroutine kịp Add.
• Gửi job và đọc kết quả trên cùng goroutine với channel không buffer: gây deadlock như đã phân tích ở mục 4.
• Không select trên ctx.Done() khi gửi kết quả: dẫn tới goroutine leak khi phía đọc đã thoát sớm.
• Chọn số worker tùy tiện: với tác vụ CPU-bound, hãy dùng runtime.NumCPU(). Với tác vụ I/O-bound, hãy dựa vào giới hạn của tài nguyên phía sau, ví dụ kích thước connection pool.

Khi nào không nên dùng worker pool:

• Số tác vụ nhỏ và đã biết trước, ví dụ gọi 3 service song song. Lúc này errgroup không cần SetLimit là đủ.
• Tác vụ cực ngắn, chỉ vài micro giây. Chi phí đi qua channel có thể lớn hơn chính công việc, nên xử lý tuần tự hoặc chia lô (batch) sẽ hiệu quả hơn.
• Bạn cần giới hạn theo tốc độ (request/giây) chứ không theo số lượng đồng thời. Trường hợp này cần rate limiter, ví dụ golang.org/x/time/rate. Worker pool không thay thế được rate limiter.

9. Kết luận

Worker pool giúp bạn kiểm soát mức song song trong Go: số tác vụ có thể rất lớn nhưng tài nguyên phía sau vẫn được bảo vệ. Bài viết đã đi qua các bước sau:

• Pool cơ bản với channel và sync.WaitGroup.
• Thu kết quả và lỗi qua channel results, đóng channel đúng chỗ.
• Hủy an toàn bằng context ở cả hai chiều gửi.
• Rút gọn bằng errgroup.SetLimit với hàm Process generic.
• Kiểm thử giới hạn đồng thời và lỗi bằng go test -race.

Với phần lớn trường hợp, errgroup.SetLimit là lựa chọn đơn giản và ít lỗi nhất. Bạn chỉ nên tự viết pool khi worker cần giữ trạng thái riêng hoặc khi cần điều khiển chi tiết luồng job và kết quả.