sync.WaitGroup, Mutex và errgroup: điều phối goroutine trong Go

1. Tổng quan

Chạy một goroutine trong Go rất dễ, chỉ cần thêm từ khóa go. Phần khó hơn là biết khi nào các goroutine xong việc, bảo vệ dữ liệu dùng chung và xử lý lỗi khi một goroutine thất bại.

Bài này đi qua ba công cụ cho ba việc đó: sync.WaitGroup, sync.Mutex và errgroup.Group. Đọc xong, bạn sẽ biết chọn công cụ nào cho từng tình huống, cách viết test chạy với -race và các lỗi hay gặp cần tránh.

Các ví dụ dùng Go 1.22 trở lên. Từ phiên bản này, mỗi vòng lặp for tạo biến mới cho từng lần lặp, nên bạn không cần viết id := id nữa.

2. Bài toán mẫu

Giả sử bạn có một danh sách mã đơn hàng (Order). Mỗi đơn cần gọi một service chậm, mất khoảng 100ms. Nếu gọi tuần tự 10 đơn, bạn mất khoảng 1 giây. Nếu gọi song song, bạn chỉ mất khoảng 100ms.

Đây là hàm giả lập việc gọi service:

package main

import (
	"fmt"
	"time"
)

func fetchOrder(id int) string {
	time.Sleep(100 * time.Millisecond)
	return fmt.Sprintf("order-%d", id)
}

Nếu bạn chỉ viết go fetchOrder(id) trong vòng lặp, hàm main sẽ kết thúc trước khi các goroutine chạy xong. Chương trình thoát và kết quả bị mất. Mục tiếp theo xử lý vấn đề này.

3. sync.WaitGroup: chờ một nhóm goroutine

3.1. Cách dùng cơ bản

WaitGroup là một bộ đếm. Bạn tăng bộ đếm bằng Add, giảm bằng Done, và Wait chặn cho tới khi bộ đếm về 0.

func main() {
	var wg sync.WaitGroup
	ids := []int{1, 2, 3}

	for _, id := range ids {
		wg.Add(1)
		go func() {
			defer wg.Done()
			fmt.Println("fetched", fetchOrder(id))
		}()
	}

	wg.Wait()
	fmt.Println("all done")
}

Code trên hoạt động như sau:

• wg.Add(1) được gọi trước khi khởi chạy goroutine, trong goroutine cha.
• defer wg.Done() bảo đảm bộ đếm luôn giảm, kể cả khi goroutine kết thúc sớm hoặc panic.
• wg.Wait() chặn main cho tới khi cả ba goroutine gọi Done.

Kết quả có thể như sau. Thứ tự các dòng fetched thay đổi giữa các lần chạy:

fetched order-2
fetched order-1
fetched order-3
all done

3.2. WaitGroup.Go trong Go 1.25

Từ Go 1.25, WaitGroup có thêm phương thức Go. Phương thức này gộp Add, go và Done thành một lời gọi:

var wg sync.WaitGroup
for _, id := range ids {
	wg.Go(func() {
		fmt.Println("fetched", fetchOrder(id))
	})
}
wg.Wait()

Viết như vậy ngắn hơn và tránh được hai lỗi phổ biến: quên gọi Done, hoặc gọi Add bên trong goroutine.

3.3. Thu kết quả mà không cần lock

Bạn thường cần giữ lại kết quả, không chỉ in ra màn hình. Một cách an toàn là tạo slice có sẵn độ dài, rồi để mỗi goroutine ghi vào một chỉ số riêng:

results := make([]string, len(ids))
var wg sync.WaitGroup
for i, id := range ids {
	wg.Go(func() {
		results[i] = fetchOrder(id)
	})
}
wg.Wait()
fmt.Println(results)

[order-1 order-2 order-3]

Không có goroutine nào ghi chung một phần tử, nên ở đây không có data race. wg.Wait() còn tạo quan hệ happens-before. Vì vậy, sau Wait, goroutine cha đọc được đầy đủ dữ liệu mà các goroutine con đã ghi.

Lưu ý: Không gọi wg.Add(1) bên trong goroutine. Nếu làm vậy, Wait() có thể chạy trước Add và trả về ngay khi công việc chưa xong.

4. sync.Mutex: bảo vệ dữ liệu dùng chung

4.1. Data race trông như thế nào

Bây giờ giả sử bạn cần cộng dồn doanh thu theo từng khách hàng (Customer), và nhiều goroutine cùng cập nhật một map:

revenue := map[string]float64{}
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
	wg.Go(func() {
		revenue["alice"] += 10
	})
}
wg.Wait()

Chương trình này có thể crash với lỗi sau:

fatal error: concurrent map writes

Trường hợp chỉ dùng một biến int còn nguy hiểm hơn. Chương trình không crash mà âm thầm cho ra kết quả sai, ví dụ 9730 thay vì 10000. Để phát hiện loại lỗi này, hãy chạy với race detector:

go run -race main.go

WARNING: DATA RACE
Write at 0x00c000012345 by goroutine 8:
...

4.2. Sửa bằng Mutex

sync.Mutex bảo đảm tại mỗi thời điểm chỉ có một goroutine chạy đoạn code nằm giữa Lock và Unlock. Cách làm phổ biến là gói mutex chung với dữ liệu mà nó bảo vệ, trong cùng một struct:

package shop

import "sync"

type RevenueTracker struct {
	mu    sync.Mutex
	total map[string]float64
}

func NewRevenueTracker() *RevenueTracker {
	return &RevenueTracker{total: make(map[string]float64)}
}

func (r *RevenueTracker) Add(customer string, amount float64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.total[customer] += amount
}

func (r *RevenueTracker) Total(customer string) float64 {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.total[customer]
}

Một vài điểm cần chú ý:

• Mọi truy cập vào total, kể cả đọc, đều phải đi qua lock.
• defer r.mu.Unlock() giúp lock luôn được nhả, kể cả khi hàm có nhiều nhánh return.
• Các method dùng receiver con trỏ (*RevenueTracker). Nếu dùng receiver giá trị, mỗi lần gọi sẽ copy cả mutex, và lock đó không còn bảo vệ được gì.

Khi lượng đọc nhiều hơn hẳn lượng ghi, bạn có thể dùng sync.RWMutex. Phương thức đọc gọi RLock/RUnlock, phương thức ghi gọi Lock/Unlock. Nhiều goroutine có thể đọc cùng lúc, còn khi ghi thì chỉ một goroutine được giữ lock.

Lưu ý: Không copy một struct chứa Mutex sau khi đã dùng nó. Lệnh go vet sẽ cảnh báo trường hợp này với thông báo copylocks. Ngoài ra, Go không có reentrant lock: nếu một goroutine gọi Lock() hai lần trên cùng một mutex, nó sẽ deadlock.

4.3. Khi nào không cần Mutex

Với một bộ đếm đơn giản, sync/atomic gọn và nhanh hơn:

var processed atomic.Int64
processed.Add(1)
fmt.Println(processed.Load())

Khi dữ liệu cần đi từ goroutine này sang goroutine khác theo dạng luồng, channel thường hợp hơn. Mutex phù hợp nhất khi bạn có một trạng thái dùng chung, như map hoặc struct, và nhiều goroutine cần đọc hoặc ghi vào đó.

5. errgroup: WaitGroup có xử lý lỗi

WaitGroup không biết gì về lỗi. Nếu fetchOrder có thể trả về error, bạn phải tự gom lỗi bằng mutex hoặc channel, và tự hủy các goroutine còn lại. Gói errgroup làm sẵn những việc này.

5.1. Thêm dependency

errgroup nằm trong module golang.org/x/sync, không thuộc thư viện chuẩn:

go get golang.org/x/sync

5.2. Gom lỗi và hủy bằng context

Bước đầu tiên là viết lại fetchOrder để hàm nhận context.Context và có thể trả về lỗi:

package shop

import (
	"context"
	"fmt"
	"time"

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

type Order struct {
	ID       int
	Customer string
}

func fetchOrder(ctx context.Context, id int) (Order, error) {
	if id < 0 {
		return Order{}, fmt.Errorf("invalid order id %d", id)
	}
	select {
	case <-time.After(100 * time.Millisecond):
		return Order{ID: id, Customer: "alice"}, nil
	case <-ctx.Done():
		return Order{}, ctx.Err()
	}
}

func LoadOrders(ctx context.Context, ids []int) ([]Order, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(3)

	orders := make([]Order, len(ids))
	for i, id := range ids {
		g.Go(func() error {
			o, err := fetchOrder(ctx, id)
			if err != nil {
				return err
			}
			orders[i] = o
			return nil
		})
	}

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

Code trên hoạt động như sau:

• errgroup.WithContext(ctx) tạo một group và một context con. Ngay khi một goroutine trả về lỗi đầu tiên, context con bị hủy.
• g.SetLimit(3) cho phép tối đa 3 goroutine chạy cùng lúc. Lần gọi g.Go tiếp theo sẽ chặn cho tới khi có chỗ trống. Nhờ vậy bạn không làm quá tải service phía sau khi danh sách có hàng nghìn mã đơn.
• g.Go nhận một hàm có kiểu func() error, và bạn không cần gọi Add/Done.
• g.Wait() chờ tất cả goroutine kết thúc rồi trả về lỗi đầu tiên. Các lỗi xảy ra sau đó bị bỏ qua.
• fetchOrder lắng nghe ctx.Done(), nên các goroutine còn lại dừng sớm thay vì chạy hết 100ms.

Gọi hàm với một mã đơn không hợp lệ:

_, err := shop.LoadOrders(context.Background(), []int{1, 2, -5, 4})
fmt.Println(err)

invalid order id -5

Nếu bạn cần chạy thử mà không chặn khi đã đạt giới hạn, hãy dùng g.TryGo(f). Hàm này trả về false thay vì chờ.

Lưu ý: Việc hủy context chỉ có tác dụng khi code bên trong goroutine chủ động kiểm tra ctx. Nếu fetchOrder dùng time.Sleep mà không lắng nghe context, goroutine vẫn chạy hết thời gian dù nhóm đã thất bại.

6. Viết test với -race

Go không có JUnit hay AssertJ. Bạn dùng gói testing của thư viện chuẩn. Điều quan trọng nhất là luôn chạy test với cờ -race, vì nhiều lỗi concurrency chỉ lộ ra khi có race detector.

package shop

import (
	"context"
	"sync"
	"testing"
)

func TestRevenueTracker_ConcurrentAdd(t *testing.T) {
	tracker := NewRevenueTracker()
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Go(func() {
			tracker.Add("alice", 10)
		})
	}
	wg.Wait()

	if got := tracker.Total("alice"); got != 10000 {
		t.Fatalf("Total(alice) = %v, want 10000", got)
	}
}

func TestLoadOrders_ReturnsAllOrders(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, 2, 3})
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	for i, o := range orders {
		if o.ID != i+1 {
			t.Errorf("orders[%d].ID = %d, want %d", i, o.ID, i+1)
		}
	}
}

func TestLoadOrders_StopsOnFirstError(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, -1, 3})
	if err == nil {
		t.Fatal("expected error, got nil")
	}
	if orders != nil {
		t.Errorf("expected nil orders, got %v", orders)
	}
}

Chạy test:

go test -race ./...

ok  	example.com/shop	0.412s

Hãy thử xóa hai dòng Lock/Unlock trong Add rồi chạy lại. Test sẽ thất bại với WARNING: DATA RACE hoặc crash vì concurrent map writes. Đây là cách nhanh để kiểm tra rằng mutex thật sự cần thiết.

Test thứ hai cũng xác nhận thứ tự kết quả được giữ đúng: orders[i] khớp với ids[i], dù các goroutine kết thúc theo thứ tự bất kỳ.

7. Lỗi hay gặp và cách chọn công cụ

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

• Gọi Add trong goroutine. Wait có thể trả về quá sớm. Hãy gọi Add trước lệnh go, hoặc dùng wg.Go.
• Truyền WaitGroup hay Mutex theo giá trị. Hàm nhận được một bản copy, và bộ đếm hoặc lock gốc không thay đổi. Hãy truyền con trỏ *sync.WaitGroup, hoặc dùng closure.
• Giữ lock trong lúc gọi I/O. Ví dụ gọi HTTP bên trong Lock. Mọi goroutine khác phải xếp hàng chờ, và chương trình mất tác dụng của việc chạy song song. Hãy chỉ giữ lock trong đoạn code ngắn nhất có thể.
• Dùng errgroup.Group rỗng (var g errgroup.Group) nhưng vẫn mong context bị hủy. Chỉ WithContext mới tạo context có khả năng hủy.
• Bỏ qua giá trị trả về của g.Wait(). Khi đó lỗi biến mất mà không ai biết.
• Khởi chạy không giới hạn goroutine cho hàng chục nghìn request ra ngoài. Hãy dùng SetLimit hoặc một worker pool.

Cách chọn công cụ:

Tình huống	Công cụ
Chờ nhiều tác vụ không trả lỗi	sync.WaitGroup
Nhiều goroutine sửa chung map hoặc struct	sync.Mutex / sync.RWMutex
Bộ đếm hoặc cờ đơn giản	sync/atomic
Tác vụ có thể lỗi, cần hủy sớm, giới hạn số goroutine	errgroup

Lưu ý: errgroup chỉ giữ lỗi đầu tiên. Nếu bạn cần biết tất cả các lỗi, ví dụ để báo cáo đơn nào hỏng, hãy để mỗi goroutine ghi lỗi vào errs[i], rồi gộp lại bằng errors.Join(errs...) sau khi gọi Wait.

8. Kết luận

sync.WaitGroup trả lời câu hỏi “khi nào xong?”. sync.Mutex trả lời câu hỏi “ai được sửa dữ liệu lúc này?”. errgroup kết hợp việc chờ với xử lý lỗi, hủy bằng context và giới hạn số goroutine chạy cùng lúc.

Trong code thực tế, khi gọi nhiều service song song, errgroup.WithContext kèm SetLimit thường là lựa chọn mặc định hợp lý. Dù dùng công cụ nào, hãy luôn chạy go test -race. Đây là cách rẻ nhất để bắt data race trước khi chúng lên production.