Worker Pools

Promise Pool vs Worker Pool

ใน TypeScript เมื่อต้องการจำกัดจำนวน concurrent operation (เช่น “เรียก API พร้อมกันไม่เกิน 3 ตัว”) คุณมักหยิบ library อย่าง p-limit มาใช้ หรือเขียน Promise pool เอง:

async function processWithPool<T>(
  items: T[],
  concurrency: number,
  processor: (item: T) => Promise<void>
) {
  const pool: Promise<void>[] = [];
  for (const item of items) {
    const p = processor(item).then(() => {
      pool.splice(pool.indexOf(p), 1);
    });
    pool.push(p);
    if (pool.length >= concurrency) {
      await Promise.race(pool);
    }
  }
  await Promise.all(pool);
}

ใน Go worker pool เป็น pattern ที่ idiomatic ใช้แค่ stdlib และอ่านง่ายกว่า: goroutine จำนวนคงที่ (workers) แต่ละตัวดึงงานจาก jobs channel ตัวเดียวกัน channel ทำหน้าที่เป็น concurrent-safe queue ส่วน close(jobs) เป็นสัญญาณว่าจะไม่มีงานเข้ามาอีก

Worker Pool Pattern พื้นฐาน

// TypeScript: p-limit library สำหรับ concurrency control
import pLimit from 'p-limit';

const limit = pLimit(3); // max 3 concurrent
const tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9];

const results = await Promise.all(
  tasks.map(n => limit(() => processTask(n)))
);

// Go: Worker Pool ด้วย goroutines + channels
package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs { // รับงานจาก jobs channel
        results <- j * j  // ส่งผลลัพธ์ไปยัง results channel
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 9

    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    var wg sync.WaitGroup
    // launch workers
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // ส่งงาน
    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs) // แจ้งว่าไม่มีงานแล้ว

    wg.Wait()
    close(results)

    sum := 0
    for r := range results {
        sum += r
    }
    fmt.Println("Sum of squares:", sum) // 285
}

Pipeline Pattern — ต่อ Workers เป็น Chain

Channel ต่อกันเป็น pipeline ได้อย่างเป็นธรรมชาติ: output channel ของ stage หนึ่งกลายเป็น input channel ของ stage ถัดไป เทียบได้กับการ chain async generator ใน JS

// TypeScript: pipeline ด้วย async generators
async function* stage1(input: AsyncIterable<number>) {
  for await (const n of input) yield n * 2;
}
async function* stage2(input: AsyncIterable<number>) {
  for await (const n of input) yield n + 1;
}

// Go: pipeline ด้วย channels
package main

import "fmt"

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

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

func main() {
    nums := generate(1, 2, 3, 4, 5)
    doubled := double(nums)
    for v := range doubled {
        fmt.Println(v) // 2, 4, 6, 8, 10
    }
}

🟦 เหตุใด Worker Pool ถึง Idiomatic ใน Go

concurrency primitives ของ Go เข้ากับ worker pool ได้อย่างเป็นธรรมชาติ:

Channel เป็น concurrent-safe queue — ไม่ต้องใช้ mutex channel จัดคิวการเข้าถึงให้เอง
for range บน channel — worker วน loop สะอาด ๆ จนกว่า channel จะถูก close โดยไม่ต้องมี done flag
close(jobs) เป็น shutdown signal — แทนที่จะส่ง sentinel value N ตัว แค่ close ครั้งเดียวก็บอกทุก worker พร้อมกัน
WaitGroup + close(results) — pool รอ worker ทุกตัวเสร็จ แล้ว close results channel จากนั้น collector ค่อยไล่อ่านได้อย่างปลอดภัย

scale ได้ง่ายมาก — เพิ่ม numWorkers เพื่อใช้ CPU core มากขึ้น หรือลดลงเพื่อ throttle การใช้ resource ภายนอก (เช่น จำกัดจำนวน database connection)

Open in Playground

package main

import (
  "fmt"
  "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
  defer wg.Done()
  for j := range jobs {
    results <- j * j
  }
}

func main() {
  const numWorkers = 3
  const numJobs = 9

  jobs := make(chan int, numJobs)
  results := make(chan int, numJobs)

  var wg sync.WaitGroup
  for w := 1; w <= numWorkers; w++ {
    wg.Add(1)
    go worker(w, jobs, results, &wg)
  }

  for j := 1; j <= numJobs; j++ {
    jobs <- j
  }
  close(jobs)

  wg.Wait()
  close(results)

  sum := 0
  for r := range results {
    sum += r
  }
  fmt.Println("Sum of squares:", sum)
}