Concurrency Patterns

สาม Pattern พื้นฐาน

Goroutines และ channels ของ Go ประกอบกันตาม pattern ที่เข้าใจดีจำนวนน้อย เมื่อคุณจำแนกได้ คุณสามารถรวมกันสร้างระบบ concurrent ที่ซับซ้อนได้

Pipeline

Pipeline คือชุด stages ที่เชื่อมต่อด้วย channels แต่ละ stage รับค่าจาก upstream channel แปลงค่า แล้วส่ง result ลง downstream channel Stages รัน concurrent — ขณะที่ stage 2 กำลัง process item N, stage 1 กำลังทำงานกับ item N+1 แล้ว

generate → square → print

Pipelines เหมาะสำหรับ stream processing ที่แต่ละ transformation เป็นอิสระ

Fan-out / Fan-in

Fan-out เริ่ม goroutines หลายตัวที่อ่านจาก input channel เดียวกัน เพื่อทำงานแบบขนาน Fan-in รวม output channels หลายตัวเป็นหนึ่งเพื่อ consumer อ่าน รวมกันทำให้ parallel processing stage ใน pipeline

Worker Pool

Worker pool คือ goroutines จำนวนคงที่ที่ consume จาก shared jobs channel ขนาด pool จำกัดการใช้ทรัพยากร (goroutines, connections, file handles) ขณะที่ยังประมวลผล concurrent

main → jobs channel → [worker 1, worker 2, worker 3] → results

ตัวอย่างที่รันได้ด้านล่างรวมทั้งสาม patterns: pipeline ที่มี generator และ squaring stage ตามด้วย worker pool ที่ประมวลผล batch of jobs แบบ deterministic

Open in Playground

package main

import (
  "fmt"
  "sync"
)

// --- Pipeline ---

func generate(nums ...int) <-chan int {
  out := make(chan int, len(nums))
  for _, n := range nums {
    out <- n
  }
  close(out)
  return out
}

func square(in <-chan int) <-chan int {
  out := make(chan int, cap(in))
  go func() {
    for n := range in {
      out <- n * n
    }
    close(out)
  }()
  return out
}

// --- Worker Pool ---

func workerPool(jobs <-chan int, results []int, wg *sync.WaitGroup) {
  for j := range jobs {
    results[j-1] = j * 2
    wg.Done()
  }
}

func main() {
  // Pipeline: 1,2,3,4,5 -> squares
  nums := generate(1, 2, 3, 4, 5)
  squares := square(nums)
  for v := range squares {
    fmt.Println(v)
  }

  // Worker pool: 3 workers, 5 jobs แต่ละงาน double index
  const numJobs = 5
  jobs := make(chan int, numJobs)
  results := make([]int, numJobs)
  var wg sync.WaitGroup

  for w := 0; w < 3; w++ {
    go workerPool(jobs, results, &wg)
  }
  for j := 1; j <= numJobs; j++ {
    wg.Add(1)
    jobs <- j
  }
  close(jobs)
  wg.Wait()

  for _, r := range results {
    fmt.Println(r)
  }
}

ข้อแลกเปลี่ยน

Pattern	ประโยชน์	ต้นทุน
worker pool	จำกัดจำนวน goroutine, reuse workers	ต้อง tune pool size, ซับซ้อนกว่า goroutine ต่อ task
fan-out / fan-in	parallel processing, aggregate results	ต้องจัดการ error collection และ cancellation
pipeline	modular stages, backpressure ง่าย	เพิ่ม latency ต่อ item (buffering overhead)
semaphore (buffered channel)	rate limit ง่าย	ต้อง drain channel เมื่อ cancel

ความเข้าใจผิดที่พบบ่อย

worker pool เสมอเร็วกว่า goroutine-per-request — ขึ้นกับ workload — I/O-bound อาจเร็วกว่าด้วย goroutine-per-request
pipeline ต้องมี fixed จำนวน stage — pipeline ใน Go flexible — stage เพิ่มลดได้แบบ dynamic
fan-out โดยไม่ limit = ดีที่สุด — ไม่มี limit ทำให้ resource exhaustion เมื่อ load สูง

💡 ตัวอย่างจากของจริง

Kubernetes ใช้ work queue + worker pool เป็น core ของ controller framework — จำกัด concurrent reconciliation

Prometheus ใช้ fan-out ต่อ scrape target และ fan-in aggregate ผลลัพธ์กลับมาที่ TSDB

Docker build ใช้ pipeline pattern สำหรับ layer building — แต่ละ stage เป็น goroutine แยกกัน

🟦 ข้อควรระวัง: goroutine leak เมื่อไม่มีใด drain channel

ถ้า goroutine บล็อกอยู่บน out <- result และ receiver ออกไปโดยไม่ drain channel goroutine นั้นจะ leak — ค้างอยู่ใน runtime ตลอดไป เปลืองหน่วยความจำและ goroutine slot วิธีแก้คือตรวจสอบให้แน่ใจว่า: (a) channel มี buffer พอสำหรับ sends ทั้งหมด, (b) receiver drain channel จนกว่า close, หรือ (c) sender ก็ select บน ctx.Done() ด้วย ใน production รัน go tool pprof บน goroutine profile เพื่อตรวจจับ leak