goroutine, channel, select, pipeline, worker pool
go f() = จ้างพ่อครัวเพิ่ม ถูกกว่า thread มาก เปิดหมื่นตัวได้:
เบื้องหลังมี Scheduler (M:P:G) + Netpoller คอยแบ่งงาน + รอเน็ตโดยไม่กินเตา เราไม่ต้องสั่งเขา แค่ไม่เขียน loop เผา CPU เปล่าๆ
unbuf := make(chan int) // ไม่มีที่พัก ส่งแล้วต้องมีคนรับทันที
buf := make(chan int, 2) // พักได้ 2 ชิ้น
ch <- 1 // ส่ง
v := <-ch // รับทิศทางบอกได้: chan<- int ส่งอย่างเดียว, <-chan int รับอย่างเดียว
func producer(ch chan<- int) {
for i := 0; i < 3; i++ { ch <- i }
close(ch) // คนส่งเท่านั้นที่ปิด!
}
func consumer(ch <-chan int) {
for v := range ch { fmt.Println(v) } // รับจน close
}ของใหม่ 1.27: ส่ง generic func เข้า channel ได้โดยไม่ต้องใส่ [T]:
นั่งเฝ้าหลายราง อันไหนมาก่อนเอาอันนั้น:
select {
case v := <-ch1:
fmt.Println(v)
case v := <-ch2:
fmt.Println(v)
case <-time.After(2 * time.Second):
fmt.Println("รอ 2 วิ เลิก!")
case <-ctx.Done():
fmt.Println("โดนยกเลิก")
}non-blocking ด้วย default:
สายพาน:
func gen(nums ...int) <-chan int {
out := make(chan int)
go func() { defer close(out); for _, n := range nums { out <- n } }()
return out
}
func sq(in <-chan int) <-chan int {
out := make(chan int)
go func() { defer close(out); for n := range in { out <- n*n } }()
return out
}
// for n := range sq(gen(1,2,3,4)) { ... }งานหนักแตกหลายคน + จำกัดพร้อมกัน:
sem := make(chan struct{}, 10)
for _, j := range jobs {
sem <- struct{}{}
wg.Add(1)
go func(job Job) {
defer wg.Done()
defer func() { <-sem }()
do(job)
}(j)
}ใช้ errgroup สั้นกว่า (ได้ limit + error + context):
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10)
for _, j := range jobs {
g.Go(func() error { return do(ctx, j) })
}
g.Wait()จำเรื่อง backpressure: อย่าปล่อย channel โตไม่จำกัด producer เร็วกว่า consumer ต้องมี buffer + drop/block อย่างตั้งใจ
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel() // ได้ cancel ต้อง defer เสมอ ไม่งั้นรั่ว
select {
case <-time.After(5 * time.Second):
fmt.Println("เสร็จ")
case <-ctx.Done():
fmt.Println(ctx.Err())
}ทุก stage ของ pipeline ต้องฟัง ctx.Done() ด้วย ไม่งั้นยกเลิกแล้วค้าง
ของใหม่ 1.27: testing/synctest + httptest.NewTestServer ไว้เทสต์ concurrent ในโลกจำลองเวลา ไม่ต้อง sleep จริง ลด flaky + runtime/pprof goroutineleak (ทดลองตั้งแต่ 1.26) ไว้จับตัวที่ลืมปิด
| อาการ | สาเหตุ | แก้ |
|---|---|---|
| deadlock! | ส่งไม่มีรับ / รับไม่มีส่ง | เช็ก go + close |
| send on closed | ส่งหลัง close | ให้คนส่งปิดที่เดียว |
| ค้างไม่จบ | ลืม close / ลืมฟัง ctx | ใส่ close + select ctx |
| race | แตะตัวเดียวกันโดยไม่มี Mutex/Channel | go run -race |
go run -race .
go test -race ./...
GODEBUG=gctrace=1 go run . # ดู GC (Green Tea ตั้งแต่ 1.26 หยุดสั้นลง)อ้างอิง: Effective Go #channels, Go Blog Share Memory By Communicating, Pipelines, Go 1.26-1.27 release notes
jobs := make(chan int, 10)
var wg sync.WaitGroup
for w := 0; w < 3; w++ { // 3 คนงานแย่งงานกันทำ
wg.Add(1)
go func() {
defer wg.Done()
for j := range jobs {
fmt.Println("ทำ", j)
}
}()
}
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs) // ปิด = บอกว่าไม่มีงานแล้ว
wg.Wait() // รอทุกคนเสร็จselect {
case v := <-ch:
fmt.Println("ได้", v)
case <-time.After(2 * time.Second):
fmt.Println("รอนานไป เลิก")
}ctx, cancel := context.WithCancel(context.Background())
go func() {
<-ctx.Done() // ใครฟังอยู่รู้พร้อมกันว่าเลิก
fmt.Println("หยุด:", ctx.Err())
}()
time.Sleep(time.Second)
cancel() // สั่งเลิกทีเดียว ทุก goroutine ที่ฟังหยุดหมดclose (ปกติคือฝั่งส่ง) อย่า close จากฝั่งรับclose แล้วอ่านต่อได้ (ได้ zero value) แต่ส่งเข้า channel ที่ปิดแล้ว = panic ทันทีdone/ctx/close) ไม่งั้นรั่ว กินแรมไปเรื่อยselect + time.After/ctx กันค้าง อย่า <-ch เปล่าๆ ในงานจริงsync.Mutex) หรือส่งผ่าน channel อย่างเดียวatomic/unsafe ถ้าไม่จำเป็น-race เสมอ ก่อนบอกว่า "เสร็จแล้ว"gen(1..10) -> square -> print ด้วย channel 3 ทอด-race แล้วแก้ data race ที่เจอ