Thread Pool Pattern #

Sebuah web server menerima ribuan request per detik. Tanpa pengendalian, setiap request akan membuat goroutine baru — dan di saat traffic tinggi, bisa ada ratusan ribu goroutine aktif sekaligus. Meskipun goroutine jauh lebih ringan dari thread OS, ratusan ribu goroutine tetap menghabiskan memori secara signifikan, menyebabkan GC pressure yang tinggi, dan bisa membuat sistem crash. Thread Pool Pattern menyelesaikan ini dengan cara yang elegan: buat sejumlah goroutine yang tetap (pool), antrekan semua pekerjaan yang masuk, dan biarkan goroutine yang ada mengambil dan mengerjakan antrian itu — tidak ada goroutine baru yang dibuat, tidak ada goroutine yang dibiarkan menganggur tanpa batas.

Apa itu Thread Pool Pattern? #

Thread Pool Pattern adalah concurrency pattern yang menyediakan sekumpulan goroutine yang sudah dibuat sebelumnya (pre-spawned) untuk mengeksekusi task yang masuk, alih-alih membuat goroutine baru untuk setiap task. Pool ini dibatasi jumlahnya sehingga konsumsi resource tetap terprediksi dan terkontrol.

Di Golang, “Thread Pool” lebih tepat disebut “Goroutine Pool” karena unit konkurensinya adalah goroutine, bukan thread OS. Namun konsep dan manfaatnya identik.

Tiga manfaat utama Thread Pool:

Resource yang terkontrol — jumlah goroutine aktif dibatasi; tidak ada goroutine explosion saat traffic tinggi
Pengurangan overhead — membuat dan menghancurkan goroutine berulang kali lebih mahal daripada menggunakannya kembali
Backpressure alami — ketika pool penuh dan antrian penuh, task baru ditolak dengan error yang bisa ditangani

flowchart LR
    subgraph "Tanpa Pool — Tak Terbatas"
        T1[Task 1] --> G1[goroutine baru]
        T2[Task 2] --> G2[goroutine baru]
        T3[Task 3] --> G3[goroutine baru]
        TN[Task N] --> GN[goroutine ke-N]
        note1["N task = N goroutine\nbisa ratusan ribu!"]
    end

    subgraph "Dengan Pool — Terkontrol"
        Q[Task Queue\nbuffered channel]
        T4[Task 1] --> Q
        T5[Task 2] --> Q
        T6[Task N] --> Q
        Q --> W1[Worker 1\ngoroutine tetap]
        Q --> W2[Worker 2\ngoroutine tetap]
        Q --> W3[Worker 3\ngoroutine tetap]
        note2["N task, hanya 3 goroutine\nterlepas dari N"]
    end

Komponen Thread Pool di Golang #

Thread Pool di Golang dibangun dari tiga primitif yang bekerja bersama.

flowchart TD
    subgraph "Thread Pool"
        direction TB
        TQ["Task Queue\nbuffered channel Task"]
        W1["Worker Goroutine 1\nfor task := range taskQueue"]
        W2["Worker Goroutine 2\nfor task := range taskQueue"]
        W3["Worker Goroutine N\nfor task := range taskQueue"]
        CTX["context.Context\nuntuk shutdown signal"]
        WG["sync.WaitGroup\ntracking worker selesai"]
    end

    TQ --> W1 & W2 & W3
    CTX -->|cancel| W1 & W2 & W3
    W1 & W2 & W3 -->|Done| WG

Komponen	Peran	Implementasi
Task Queue	Buffer antrian task yang menunggu	Buffered channel `chan Task`
Worker Goroutines	Goroutine yang mengambil dan mengeksekusi task	`go func()` yang loop `for range`
Context	Signal untuk graceful shutdown	`context.WithCancel`
WaitGroup	Tracking semua worker sudah selesai	`sync.WaitGroup`

Implementasi Lengkap: General-Purpose Thread Pool #

Task Interface dan Pool #

package pool

import (
    "context"
    "errors"
    "fmt"
    "log/slog"
    "sync"
    "sync/atomic"
    "time"
)

// Task merepresentasikan pekerjaan yang bisa dieksekusi oleh pool.
type Task interface {
    Execute(ctx context.Context) error
    TaskID() string
}

// TaskResult menyimpan hasil eksekusi sebuah task.
type TaskResult struct {
    TaskID    string
    Error     error
    Duration  time.Duration
    WorkerID  int
}

// PoolConfig menyimpan konfigurasi thread pool.
type PoolConfig struct {
    Workers     int           // jumlah goroutine worker
    QueueSize   int           // kapasitas antrian task
    TaskTimeout time.Duration // timeout per task (0 = tidak ada timeout)
    OnError     func(result TaskResult) // callback saat task gagal
    Logger      *slog.Logger
}

// DefaultPoolConfig mengembalikan konfigurasi default yang masuk akal.
func DefaultPoolConfig() PoolConfig {
    return PoolConfig{
        Workers:     10,
        QueueSize:   100,
        TaskTimeout: 30 * time.Second,
        Logger:      slog.Default(),
    }
}

// Metrics menyimpan statistik pool yang bisa dimonitor.
type Metrics struct {
    TasksSubmitted atomic.Int64
    TasksCompleted atomic.Int64
    TasksFailed    atomic.Int64
    TasksRejected  atomic.Int64
    ActiveWorkers  atomic.Int32
}

// Pool adalah Thread Pool yang thread-safe.
type Pool struct {
    config  PoolConfig
    tasks   chan Task
    results chan TaskResult
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc
    metrics Metrics
    started bool
    mu      sync.Mutex
}

// NewPool membuat pool baru dengan konfigurasi yang diberikan.
func NewPool(config PoolConfig) *Pool {
    if config.Workers <= 0 {
        config.Workers = 1
    }
    if config.QueueSize <= 0 {
        config.QueueSize = config.Workers * 10
    }

    ctx, cancel := context.WithCancel(context.Background())
    return &Pool{
        config:  config,
        tasks:   make(chan Task, config.QueueSize),
        results: make(chan TaskResult, config.QueueSize),
        ctx:     ctx,
        cancel:  cancel,
    }
}

// Start memulai semua worker goroutine.
func (p *Pool) Start() {
    p.mu.Lock()
    defer p.mu.Unlock()

    if p.started {
        return
    }
    p.started = true

    for i := 0; i < p.config.Workers; i++ {
        workerID := i + 1
        p.wg.Add(1)
        go p.runWorker(workerID)
    }

    p.config.Logger.Info("thread pool started",
        "workers", p.config.Workers,
        "queue_size", p.config.QueueSize,
    )
}

// Submit menambahkan task ke antrian pool.
// Mengembalikan error jika pool sudah shutdown atau antrian penuh.
func (p *Pool) Submit(task Task) error {
    select {
    case <-p.ctx.Done():
        return errors.New("pool is shutting down")
    default:
    }

    select {
    case p.tasks <- task:
        p.metrics.TasksSubmitted.Add(1)
        return nil
    default:
        p.metrics.TasksRejected.Add(1)
        return fmt.Errorf("task queue is full (capacity: %d)", p.config.QueueSize)
    }
}

// SubmitWait menambahkan task dan menunggu hingga ada slot tersedia.
// Mengembalikan error jika pool shutdown sebelum task bisa masuk.
func (p *Pool) SubmitWait(task Task) error {
    select {
    case <-p.ctx.Done():
        return errors.New("pool is shutting down")
    case p.tasks <- task:
        p.metrics.TasksSubmitted.Add(1)
        return nil
    }
}

// Results mengembalikan channel untuk membaca hasil task.
func (p *Pool) Results() <-chan TaskResult {
    return p.results
}

// Stop menghentikan pool secara graceful.
// Menunggu semua worker selesai memproses task yang sedang dikerjakan.
func (p *Pool) Stop() {
    p.cancel()          // signal semua worker untuk berhenti
    close(p.tasks)      // tutup channel — worker akan drain sisa task lalu berhenti
    p.wg.Wait()         // tunggu semua worker selesai
    close(p.results)    // tutup results channel setelah semua worker berhenti
    p.config.Logger.Info("thread pool stopped",
        "submitted", p.metrics.TasksSubmitted.Load(),
        "completed", p.metrics.TasksCompleted.Load(),
        "failed", p.metrics.TasksFailed.Load(),
        "rejected", p.metrics.TasksRejected.Load(),
    )
}

// Metrics mengembalikan snapshot statistik pool saat ini.
func (p *Pool) GetMetrics() map[string]int64 {
    return map[string]int64{
        "submitted":      p.metrics.TasksSubmitted.Load(),
        "completed":      p.metrics.TasksCompleted.Load(),
        "failed":         p.metrics.TasksFailed.Load(),
        "rejected":       p.metrics.TasksRejected.Load(),
        "active_workers": int64(p.metrics.ActiveWorkers.Load()),
        "queue_length":   int64(len(p.tasks)),
    }
}

// runWorker adalah goroutine yang terus mengambil dan mengeksekusi task.
func (p *Pool) runWorker(workerID int) {
    defer p.wg.Done()
    defer p.metrics.ActiveWorkers.Add(-1)
    p.metrics.ActiveWorkers.Add(1)

    p.config.Logger.Debug("worker started", "worker_id", workerID)

    for task := range p.tasks {
        p.executeTask(workerID, task)
    }

    p.config.Logger.Debug("worker stopped", "worker_id", workerID)
}

// executeTask mengeksekusi satu task dengan timeout dan panic recovery.
func (p *Pool) executeTask(workerID int, task Task) {
    start := time.Now()

    // Buat context dengan timeout jika dikonfigurasi
    ctx := p.ctx
    var cancel context.CancelFunc
    if p.config.TaskTimeout > 0 {
        ctx, cancel = context.WithTimeout(p.ctx, p.config.TaskTimeout)
        defer cancel()
    }

    // Eksekusi task dengan panic recovery
    var execErr error
    func() {
        defer func() {
            if r := recover(); r != nil {
                execErr = fmt.Errorf("task panicked: %v", r)
            }
        }()
        execErr = task.Execute(ctx)
    }()

    result := TaskResult{
        TaskID:   task.TaskID(),
        Error:    execErr,
        Duration: time.Since(start),
        WorkerID: workerID,
    }

    if execErr != nil {
        p.metrics.TasksFailed.Add(1)
        if p.config.OnError != nil {
            p.config.OnError(result)
        }
    } else {
        p.metrics.TasksCompleted.Add(1)
    }

    // Kirim result ke channel (non-blocking — jangan block worker)
    select {
    case p.results <- result:
    default:
        // Results buffer penuh — log tapi jangan block worker
        p.config.Logger.Warn("results buffer full, dropping result",
            "task_id", task.TaskID())
    }
}

Concrete Tasks #

package pool

import (
    "context"
    "fmt"
    "time"
)

// HTTPFetchTask mengambil URL dan menyimpan hasilnya.
type HTTPFetchTask struct {
    id      string
    url     string
    client  *http.Client
    result  *[]byte
}

func NewHTTPFetchTask(id, url string, client *http.Client, result *[]byte) Task {
    return &HTTPFetchTask{id: id, url: url, client: client, result: result}
}

func (t *HTTPFetchTask) TaskID() string { return t.id }

func (t *HTTPFetchTask) Execute(ctx context.Context) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.url, nil)
    if err != nil {
        return fmt.Errorf("failed to build request: %w", err)
    }

    resp, err := t.client.Do(req)
    if err != nil {
        return fmt.Errorf("request failed: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("unexpected status: %d", resp.StatusCode)
    }

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return fmt.Errorf("read body failed: %w", err)
    }

    *t.result = body
    return nil
}


// EmailTask mengirim satu email.
type EmailTask struct {
    id        string
    recipient string
    subject   string
    body      string
    mailer    EmailSender
}

func NewEmailTask(id, recipient, subject, body string, mailer EmailSender) Task {
    return &EmailTask{
        id: id, recipient: recipient,
        subject: subject, body: body, mailer: mailer,
    }
}

func (t *EmailTask) TaskID() string { return t.id }

func (t *EmailTask) Execute(ctx context.Context) error {
    return t.mailer.Send(ctx, t.recipient, t.subject, t.body)
}


// FuncTask memungkinkan penggunaan closure sebagai task — sangat fleksibel.
type FuncTask struct {
    id string
    fn func(ctx context.Context) error
}

func NewFuncTask(id string, fn func(ctx context.Context) error) Task {
    return &FuncTask{id: id, fn: fn}
}

func (t *FuncTask) TaskID() string { return t.id }

func (t *FuncTask) Execute(ctx context.Context) error {
    return t.fn(ctx)
}

Demonstrasi: Batch URL Fetching #

func fetchURLsConcurrently(urls []string) map[string][]byte {
    config := pool.PoolConfig{
        Workers:     5,
        QueueSize:   len(urls),
        TaskTimeout: 10 * time.Second,
        OnError: func(result pool.TaskResult) {
            log.Printf("Task %s failed after %v: %v",
                result.TaskID, result.Duration, result.Error)
        },
    }

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    results := make(map[string][]byte, len(urls))
    resultBytes := make([][]byte, len(urls))
    var mu sync.Mutex

    // Submit semua task
    for i, url := range urls {
        idx := i
        urlCopy := url
        task := pool.NewFuncTask(fmt.Sprintf("fetch-%d", idx), func(ctx context.Context) error {
            // ... fetch logic
            mu.Lock()
            results[urlCopy] = resultBytes[idx]
            mu.Unlock()
            return nil
        })
        if err := p.Submit(task); err != nil {
            log.Printf("Cannot submit task for %s: %v", url, err)
        }
    }

    // Baca hasil
    completed := 0
    for result := range p.Results() {
        completed++
        if completed >= len(urls) {
            break
        }
    }

    return results
}

Graceful Shutdown: Dua Strategi #

Graceful shutdown adalah aspek paling kritis dari Thread Pool. Ada dua pendekatan yang berbeda tergantung kebutuhan.

sequenceDiagram
    participant C as Client
    participant P as Pool
    participant W as Workers

    Note over C,W: Strategi 1: Drain then Stop

    C->>P: Stop()
    P->>P: cancel() — signal berhenti terima task baru
    P->>P: close(tasks) — tutup channel task
    W->>W: drain sisa task dalam channel
    W->>W: selesai → exit
    P->>C: wg.Wait() selesai

    Note over C,W: Strategi 2: Hard Stop (abandon sisa task)
    C->>P: ForceStop()
    P->>P: cancel() — context cancelled
    W->>W: ctx.Done() → exit segera
    Note over W: Sisa task dalam queue diabaikan

// GracefulPool mendukung dua mode shutdown.
type GracefulPool struct {
    Pool
}

// Stop menyelesaikan semua task yang sudah ada dalam queue sebelum berhenti.
// Tidak menerima task baru setelah Stop dipanggil.
func (p *GracefulPool) Stop() {
    p.cancel()     // stop menerima task baru
    close(p.tasks) // signal worker untuk drain dan exit
    p.wg.Wait()    // tunggu semua selesai
}

// ForceStop menghentikan semua worker segera, mengabaikan sisa task dalam queue.
func (p *GracefulPool) ForceStop() {
    p.cancel() // cancel context — worker akan exit saat cek ctx.Done()
    // tidak menutup tasks channel — worker keluar dari ctx.Done() bukan dari range
    p.wg.Wait()
}

// Contoh worker yang mendukung kedua mode:
func (p *Pool) runWorkerWithCancel(workerID int) {
    defer p.wg.Done()

    for {
        select {
        case task, ok := <-p.tasks:
            if !ok {
                // Channel ditutup — graceful shutdown: semua task sudah di-drain
                return
            }
            p.executeTask(workerID, task)

        case <-p.ctx.Done():
            // Context cancelled — force shutdown: drain sisa tanpa execute
            return
        }
    }
}

Dynamic Pool Sizing #

Untuk beban kerja yang tidak merata, ukuran pool bisa disesuaikan secara dinamis.

// DynamicPool menyesuaikan jumlah worker berdasarkan panjang antrian.
type DynamicPool struct {
    Pool
    minWorkers int
    maxWorkers int
    scaleUpAt  int // scale up jika queue lebih dari N%
    mu         sync.Mutex
    workerCount int
}

func NewDynamicPool(min, max int, queueSize int) *DynamicPool {
    p := &DynamicPool{
        minWorkers: min,
        maxWorkers: max,
        scaleUpAt:  70, // scale up jika antrian > 70% kapasitas
    }
    p.config.Workers = min
    p.config.QueueSize = queueSize
    return p
}

// AutoScale memeriksa kondisi pool dan menambah/mengurangi worker.
// Dipanggil secara periodik oleh background goroutine.
func (p *DynamicPool) AutoScale() {
    queueUsage := float64(len(p.tasks)) / float64(cap(p.tasks)) * 100

    p.mu.Lock()
    defer p.mu.Unlock()

    if queueUsage > float64(p.scaleUpAt) && p.workerCount < p.maxWorkers {
        // Scale up: tambah satu worker
        p.workerCount++
        p.wg.Add(1)
        go p.runWorker(p.workerCount)
        p.config.Logger.Info("scaled up", "workers", p.workerCount, "queue_usage", queueUsage)
    }
    // Scale down: lebih kompleks, butuh sinyal ke worker tertentu untuk exit
}

// StartAutoScaler menjalankan background goroutine yang mengecek load secara periodik.
func (p *DynamicPool) StartAutoScaler(interval time.Duration) {
    go func() {
        ticker := time.NewTicker(interval)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                p.AutoScale()
            case <-p.ctx.Done():
                return
            }
        }
    }()
}

Monitoring Pool dengan Prometheus #

Untuk production, pool metrics perlu diekspor ke monitoring system.

// PoolMonitor mengekspos metrics pool sebagai Prometheus gauge.
type PoolMonitor struct {
    pool           *Pool
    queueGauge     prometheus.Gauge
    workersGauge   prometheus.Gauge
    completedCount prometheus.Counter
    failedCount    prometheus.Counter
}

func NewPoolMonitor(pool *Pool, namespace string) *PoolMonitor {
    return &PoolMonitor{
        pool: pool,
        queueGauge: prometheus.NewGauge(prometheus.GaugeOpts{
            Namespace: namespace,
            Name:      "thread_pool_queue_length",
            Help:      "Current number of tasks waiting in queue",
        }),
        workersGauge: prometheus.NewGauge(prometheus.GaugeOpts{
            Namespace: namespace,
            Name:      "thread_pool_active_workers",
            Help:      "Current number of active workers",
        }),
    }
}

// Collect dipanggil oleh Prometheus scraper untuk mengambil metrics saat ini.
func (m *PoolMonitor) Collect() {
    metrics := m.pool.GetMetrics()
    m.queueGauge.Set(float64(metrics["queue_length"]))
    m.workersGauge.Set(float64(metrics["active_workers"]))
}

Testing Thread Pool #

func TestPool_ExecutesAllTasks(t *testing.T) {
    config := pool.DefaultPoolConfig()
    config.Workers = 3
    config.QueueSize = 20

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    const taskCount = 10
    var completed atomic.Int32

    for i := 0; i < taskCount; i++ {
        taskID := fmt.Sprintf("task-%d", i)
        task := pool.NewFuncTask(taskID, func(ctx context.Context) error {
            completed.Add(1)
            return nil
        })
        if err := p.Submit(task); err != nil {
            t.Fatalf("submit failed: %v", err)
        }
    }

    // Kumpulkan semua hasil
    received := 0
    for result := range p.Results() {
        if result.Error != nil {
            t.Errorf("task %s failed: %v", result.TaskID, result.Error)
        }
        received++
        if received >= taskCount {
            break
        }
    }

    if int(completed.Load()) != taskCount {
        t.Errorf("expected %d tasks completed, got %d", taskCount, completed.Load())
    }
}

func TestPool_RejectsWhenFull(t *testing.T) {
    config := pool.PoolConfig{
        Workers:   1,
        QueueSize: 2, // sangat kecil untuk memudahkan testing
    }
    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    // Isi antrian dengan task yang lambat
    slowTask := pool.NewFuncTask("slow", func(ctx context.Context) error {
        time.Sleep(1 * time.Second)
        return nil
    })

    for i := 0; i < 5; i++ {
        _ = p.Submit(slowTask)
    }

    // Task berikutnya harus ditolak
    err := p.Submit(slowTask)
    if err == nil {
        t.Error("expected error when queue is full")
    }
}

func TestPool_PanicRecovery(t *testing.T) {
    config := pool.DefaultPoolConfig()
    config.Workers = 2

    p := pool.NewPool(config)
    p.Start()
    defer p.Stop()

    // Task yang panic
    panicTask := pool.NewFuncTask("panic-task", func(ctx context.Context) error {
        panic("intentional panic for testing")
    })
    normalTask := pool.NewFuncTask("normal-task", func(ctx context.Context) error {
        return nil
    })

    _ = p.Submit(panicTask)
    _ = p.Submit(normalTask)

    var panicResult, normalResult pool.TaskResult
    received := 0
    for result := range p.Results() {
        if result.TaskID == "panic-task" {
            panicResult = result
        } else {
            normalResult = result
        }
        received++
        if received >= 2 {
            break
        }
    }

    // Panic task harus menghasilkan error, bukan crash pool
    if panicResult.Error == nil {
        t.Error("expected error from panicking task")
    }

    // Task normal harus tetap berjalan setelah ada panic
    if normalResult.Error != nil {
        t.Errorf("normal task should succeed: %v", normalResult.Error)
    }
}

func TestPool_GracefulShutdown(t *testing.T) {
    config := pool.PoolConfig{Workers: 2, QueueSize: 10}
    p := pool.NewPool(config)
    p.Start()

    var completed atomic.Int32
    for i := 0; i < 5; i++ {
        _ = p.Submit(pool.NewFuncTask(fmt.Sprintf("task-%d", i),
            func(ctx context.Context) error {
                time.Sleep(10 * time.Millisecond)
                completed.Add(1)
                return nil
            }))
    }

    // Stop harus menunggu semua task selesai
    p.Stop()

    if completed.Load() != 5 {
        t.Errorf("graceful shutdown should complete all queued tasks, got %d/5",
            completed.Load())
    }
}

Thread Pool vs Worker Pool #

Dua pattern yang sering dianggap sama padahal ada perbedaan konseptual.

Aspek	Thread Pool	Worker Pool
Fokus	Mengelola lifecycle goroutine (create, reuse, destroy)	Mendistribusikan task ke worker yang spesifik
Task routing	Task dipilih worker mana saja yang tersedia	Task bisa diarahkan ke worker tertentu
Worker identity	Worker anonim dan interchaneable	Worker bisa punya state sendiri
Use case	HTTP request handling, batch processing	Worker dengan koneksi DB sendiri, shard-based processing
Kompleksitas	Lebih sederhana	Lebih fleksibel tapi lebih kompleks

Kapan Menggunakan dan Kapan Tidak #

GUNAKAN Thread Pool jika:
  ✓ Ada banyak task independen yang perlu dieksekusi secara concurrent
  ✓ Perlu membatasi jumlah goroutine aktif untuk mengendalikan resource
  ✓ Task datang lebih cepat dari kemampuan satu goroutine memprosesnya
  ✓ Perlu backpressure — tolak task baru ketika sistem overloaded
  ✓ Task bersifat short-lived dan homogen

HINDARI Thread Pool jika:
  ✗ Hanya ada sedikit task yang jarang — goroutine biasa sudah cukup
  ✗ Task perlu mengakses state tertentu yang hanya dimiliki satu worker
  ✗ Task memerlukan ordering yang dijamin — gunakan single goroutine + channel
  ✗ Semua task berjalan sangat lama — pool akan cepat habis; better pakai Worker Pool

Checklist Review Thread Pool #

KONFIGURASI:
  □ Jumlah worker ditentukan berdasarkan profiling, bukan tebakan
  □ Ukuran antrian cukup besar untuk menyerap burst tapi tidak tak terbatas
  □ Task timeout dikonfigurasi untuk mencegah worker terjebak forever
  □ OnError callback dikonfigurasi untuk logging atau alerting

SHUTDOWN:
  □ Stop() dipanggil — tidak ada goroutine yang bocor
  □ Defer p.Stop() digunakan di main atau test untuk memastikan cleanup
  □ Graceful vs force shutdown dipilih sesuai kebutuhan
  □ Results channel di-drain atau ditutup dengan benar

PANIC SAFETY:
  □ Setiap task execution dibungkus dengan recover()
  □ Panic di satu task tidak crash seluruh pool
  □ Error dari panic di-wrap sebagai error dan dikembalikan sebagai TaskResult

MONITORING:
  □ Queue length dimonitor — alert jika mendekati kapasitas
  □ Active worker count dimonitor
  □ Task failure rate dimonitor

Ringkasan #

Thread Pool membatasi jumlah goroutine aktif — resource terkelola, tidak ada goroutine explosion saat traffic tinggi, dan GC pressure berkurang.

Tiga komponen kunci: task queue (buffered channel), worker goroutines (loop for range), dan context untuk shutdown signal.

Buffered channel adalah task queue — ukurannya menentukan seberapa banyak task bisa antri sebelum Submit mulai menolak; pilih dengan bijak.

Panic recovery wajib — satu task yang panic tidak boleh crash worker; bungkus setiap Execute dengan defer recover().

Graceful shutdown: cancel() + close(tasks) + wg.Wait() — urutan ini memastikan semua task dalam antrian diproses sebelum worker berhenti.

Results channel memungkinkan non-blocking result collection — worker tidak perlu menunggu consumer; gunakan buffer yang cukup.

Dynamic sizing untuk beban tidak merata — scale up ketika antrian > 70% kapasitas, scale down saat idle.

Monitor queue length — jika antrian sering penuh, pool perlu lebih banyak worker atau task perlu dipercepat; jika selalu kosong, pool bisa dikecilkan.

← Sebelumnya: Interpreter Berikutnya: Worker Pool →