WaitGroup trong Golang: Add, Done, Wait và worker pool

Mẫu cơ bản

sync.WaitGroup đếm số goroutine đang chạy. Add tăng bộ đếm, Done giảm nó, Wait chặn cho đến khi bộ đếm về 0.

package main

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

func download(name string) int {
	time.Sleep(10 * time.Millisecond) // pretend to work
	return len(name) * 100
}

func main() {
	files := []string{"a.txt", "report.pdf", "photo.jpg"}
	sizes := make([]int, len(files))

	var wg sync.WaitGroup
	for i, f := range files {
		wg.Add(1) // before the go statement
		go func() {
			defer wg.Done() // runs even if download panics
			sizes[i] = download(f)
		}()
	}
	wg.Wait()

	for i, f := range files {
		fmt.Printf("%-11s %d bytes\n", f, sizes[i])
	}
}

Ba lượt tải chạy đồng thời, nên chương trình mất khoảng 10 ms thay vì 30. Kết quả được in theo đúng thứ tự đầu vào, vì mỗi goroutine chỉ ghi vào chỉ số của riêng nó trong sizes, và main chỉ đọc chúng sau Wait.

Giá trị zero của WaitGroup dùng được ngay. Không cần hàm khởi tạo.

Ba quy tắc

Gọi Add trước go, không gọi bên trong goroutine. Nếu goroutine tự gọi Add, main có thể tới Wait trước khi goroutine nào kịp bắt đầu, thấy bộ đếm bằng 0 và trả về trong khi công việc còn chưa khởi động. Khi đã biết trước số lượng, gọi wg.Add(len(files)) một lần trước vòng lặp cũng tương đương.

Gọi Done bằng defer ở dòng đầu tiên của goroutine. Goroutine nào trả về sớm vì lỗi, hay bị panic, vẫn giảm được bộ đếm. Thiếu một Done là Wait bị chặn mãi mãi. Nếu đó là goroutine cuối cùng còn lại, runtime báo fatal error: all goroutines are asleep với sync.WaitGroup.Wait trong trace.

Không bao giờ sao chép WaitGroup sau lần dùng đầu tiên. Hãy truyền *sync.WaitGroup vào hàm, hoặc bắt biến trong một closure như ví dụ trên.

Truyền WaitGroup vào hàm

Khi thân goroutine là một hàm có tên, hãy truyền con trỏ:

package main

import (
	"fmt"
	"sync"
)

func worker(id int, wg *sync.WaitGroup, out []string) {
	defer wg.Done()
	out[id] = fmt.Sprintf("worker %d done", id)
}

func main() {
	var wg sync.WaitGroup
	out := make([]string, 3)

	for id := range 3 {
		wg.Add(1)
		go worker(id, &wg, out)
	}
	wg.Wait()

	for _, line := range out {
		fmt.Println(line)
	}
}

Nếu wg sync.WaitGroup là tham số kiểu giá trị, mỗi worker sẽ gọi Done trên bản sao của riêng nó và main sẽ bị chặn trong Wait mãi mãi. go vet bắt được lỗi này trước khi bạn chạy bất cứ thứ gì:

./main.go:8:24: worker passes lock by value: sync.WaitGroup contains sync.noCopy

Một thiết kế gọn hơn là để worker không dính gì đến xử lý đồng thời: cho nó là một hàm bình thường và làm phần sổ sách Add/Done trong closure của nơi gọi. Khi đó worker dễ test và gọi đồng bộ được.

Bộ đếm âm

Done chính là Add(-1). Nếu bộ đếm xuống dưới 0, chương trình panic:

package main

import (
	"fmt"
	"sync"
)

func main() {
	defer func() {
		fmt.Println("recovered:", recover())
	}()

	var wg sync.WaitGroup
	wg.Add(1)
	wg.Done()
	wg.Done() // one Done too many
}

Đầu ra là recovered: sync: negative WaitGroup counter. Nguyên nhân thường gặp là một goroutine có defer wg.Done() nhưng ở nhánh nào đó lại gọi thêm wg.Done() trực tiếp.

Gom lỗi

WaitGroup chỉ biết đếm. Với lỗi, hãy cho mỗi goroutine một ô riêng và kiểm tra chúng sau Wait:

package main

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

func check(host string) error {
	if host == "db" {
		return errors.New("db: connection refused")
	}
	return nil
}

func main() {
	hosts := []string{"web", "db", "cache"}
	errs := make([]error, len(hosts))

	var wg sync.WaitGroup
	for i, h := range hosts {
		wg.Add(1)
		go func() {
			defer wg.Done()
			errs[i] = check(h)
		}()
	}
	wg.Wait()

	if err := errors.Join(errs...); err != nil {
		fmt.Println("health check failed:", err)
		return
	}
	fmt.Println("all healthy")
}

errors.Join (Go 1.20) bỏ qua các giá trị nil và trả về nil nếu tất cả đều là nil, nên nó gộp "mỗi goroutine một lỗi" mà không cần sổ sách gì thêm.

Nếu bạn muốn dừng phần việc còn lại ngay khi một goroutine thất bại, hãy dùng golang.org/x/sync/errgroup. Nó là một WaitGroup cộng với lỗi đầu tiên cộng với một context bị hủy khi có thất bại, và g.SetLimit(n) giới hạn mức đồng thời. Package này nằm ngoài thư viện chuẩn, nên không chạy được trong trình soạn thảo của trang này:

g, ctx := errgroup.WithContext(ctx)
for _, h := range hosts {
	g.Go(func() error { return checkCtx(ctx, h) })
}
if err := g.Wait(); err != nil {
	return err // the first error; ctx was cancelled for the others
}

Worker pool

Một số goroutine cố định đọc việc từ một channel giúp giữ mức đồng thời trong giới hạn, dù có bao nhiêu việc đi nữa. WaitGroup cho bạn biết khi nào mọi worker đã xong, và đó là lúc có thể đóng channel kết quả.

package main

import (
	"fmt"
	"sort"
	"sync"
)

type result struct {
	job, value, worker int
}

func main() {
	jobs := make(chan int)
	results := make(chan result)

	const workers = 3
	var wg sync.WaitGroup
	for w := 1; w <= workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs { // ends when jobs is closed
				results <- result{job: j, value: j * j, worker: w}
			}
		}()
	}

	// Close results once every worker has exited.
	go func() {
		wg.Wait()
		close(results)
	}()

	// Feed the jobs, then close the channel so workers stop.
	go func() {
		for j := 1; j <= 8; j++ {
			jobs <- j
		}
		close(jobs)
	}()

	var all []result
	for r := range results {
		all = append(all, r)
	}
	sort.Slice(all, func(a, b int) bool { return all[a].job < all[b].job })

	for _, r := range all {
		fmt.Printf("job %d -> %d\n", r.job, r.value)
	}
	fmt.Println("jobs done:", len(all))
}

Thứ tự của ba phần này rất quan trọng:

• main phải nhận kết quả trong khi các worker đang chạy. Nếu main gọi thẳng wg.Wait() trước khi đọc, các worker sẽ bị chặn khi gửi vào results, không bao giờ tới được Done, và mọi thứ rơi vào deadlock. Vì thế Wait chạy trong một goroutine riêng.
• close(results) chỉ xảy ra sau Wait, nên không worker nào gửi vào một channel đã đóng.
• Phần cấp việc cũng chạy trong một goroutine, nên việc cấp việc và gom kết quả diễn ra song song.

Worker nào xử lý việc nào thay đổi theo từng lần chạy, nên chương trình sắp xếp theo việc trước khi in. Mọi thứ nó in ra đều cố định.

WaitGroup, channel hay errgroup

Go 1.25 bổ sung wg.Go(func() { ... }), tự làm Add(1) và Done có defer giúp bạn. Code cho Go 1.24 trở về trước, kể cả trình soạn thảo của trang này, dùng dạng tường minh như ở trên.

Lỗi thường gặp

• Gọi wg.Add(1) bên trong goroutine. Wait có thể trả về trước khi nó kịp chạy.
• Quên Done khi trả về sớm. Luôn viết defer wg.Done().
• Truyền WaitGroup theo giá trị. Hãy dùng con trỏ; go vet sẽ đánh dấu bản sao.
• Chờ trong chính goroutine phải đọc hết một channel. Chuyển wg.Wait() cùng close sang một goroutine riêng.
• Dùng lại WaitGroup khi Wait trước đó chưa trả về. Chỉ bắt đầu một vòng gọi Add mới sau khi Wait đã xong.