Worker Pool Pattern #

Setiap sistem backend yang memproses banyak pekerjaan secara paralel pasti menghadapi pertanyaan yang sama: berapa banyak worker yang harus dibuat? Terlalu sedikit, dan sistem menjadi lambat karena antrian job terus menumpuk. Terlalu banyak, dan sistem justru kolaps karena resource contention — CPU bersaing, koneksi database habis, dan memory meledak. Worker Pool Pattern hadir untuk menjawab ketegangan ini secara elegan. Alih-alih membuat satu goroutine per task (yang berisiko tak terkendali), pattern ini mendefinisikan jumlah worker yang terkontrol, mempertemukan mereka dengan job queue, dan membiarkan sistem berjalan stabil dalam jangka panjang.

Apa itu Worker Pool Pattern? #

Worker Pool Pattern adalah pola concurrency di mana sekumpulan worker — jumlahnya tetap atau terkontrol — dibuat terlebih dahulu dan siap memproses job yang masuk melalui sebuah antrian bersama. Worker tidak dibuat per task; sebaliknya, mereka hidup sepanjang siklus aplikasi dan mengambil job dari queue satu per satu.

Tiga komponen utama membentuk pattern ini:

Secara struktural, alur kerja Worker Pool terlihat seperti ini:

flowchart LR
    P[Producer / Dispatcher] --> Q[(Job Queue)]
    Q --> W1[Worker 1]
    Q --> W2[Worker 2]
    Q --> W3[Worker N]
    W1 --> R[(Result / Output)]
    W2 --> R
    W3 --> R

Yang membedakan Worker Pool dari pendekatan naif adalah reuse. Worker tidak dibuat dan dihancurkan setiap kali ada job — mereka sudah ada, menunggu, dan langsung bekerja begitu job tiba. Ini menghilangkan overhead pembuatan goroutine berulang dan menjaga memory footprint tetap terprediksi.

Mengapa Pattern Ini Penting? #

Sebelum memahami cara implementasinya, penting untuk mengerti mengapa Worker Pool dibutuhkan — dan masalah apa yang muncul jika kamu tidak menggunakannya.

Masalah Tanpa Worker Pool #

Bayangkan sebuah HTTP handler yang memproses setiap request dengan goroutine baru:

// ANTI-PATTERN: membuat goroutine baru untuk setiap job tanpa batas
http.HandleFunc("/process", func(w http.ResponseWriter, r *http.Request) {
    go processJob(r.Context()) // tidak ada kontrol jumlah goroutine
})

// BENAR: kirim job ke worker pool yang sudah berjalan
http.HandleFunc("/process", func(w http.ResponseWriter, r *http.Request) {
    jobQueue <- Job{ctx: r.Context()}
})

Pendekatan pertama tampak sederhana, tapi dalam kondisi traffic tinggi ia bisa menciptakan ribuan goroutine secara bersamaan. Masing-masing goroutine mengonsumsi memory, bersaing memperebutkan CPU, dan bisa membanjiri resource eksternal seperti database atau API pihak ketiga.

Worker Pool mengatasi ini dengan menetapkan batas atas yang jelas: tidak peduli berapa banyak job yang datang, hanya N worker yang berjalan bersamaan.

Empat Tujuan Utama #

flowchart TD
    WP[Worker Pool Pattern]
    WP --> A[Kontrol Concurrency]
    WP --> B[Resource Efficiency]
    WP --> C[Stabilitas Sistem]
    WP --> D[Throughput Konsisten]

    A --> A1["Batasi goroutine yang berjalan bersamaan"]
    B --> B1["Worker di-reuse, tidak dibuat ulang setiap task"]
    C --> C1["Cegah thundering herd problem"]
    D --> D1["Proses job secara steady, bukan spike ekstrem"]

Kontrol Concurrency — Jumlah worker yang berjalan bersamaan bersifat deterministik. Kamu tahu persis berapa banyak task yang diproses secara paralel pada satu waktu.

Resource Efficiency — Pembuatan goroutine memang murah di Go, tapi tidak gratis. Worker Pool menghilangkan overhead pembuatan berulang dan memastikan stack goroutine hanya dialokasikan sekali.

Stabilitas Sistem — Thundering herd problem terjadi ketika banyak permintaan serentak menyerbu resource yang sama. Worker Pool menyerap lonjakan ini melalui job queue, membiarkan worker memproses secara berurutan.

Throughput Konsisten — Sistem dengan Worker Pool berperilaku lebih dapat diprediksi di bawah beban karena tidak ada spike mendadak yang mengganggu latency.

Kapan Menggunakan Worker Pool? #

Tidak semua situasi membutuhkan Worker Pool. Pattern ini paling efektif pada kondisi tertentu.

flowchart TD
    S{Apakah task-nya independen?} -- Tidak --> X1[❌ Tidak cocok — pertimbangkan Pipeline]
    S -- Ya --> T{Apakah task bersifat I/O bound?}
    T -- Ya --> U[✅ Worker Pool sangat cocok]
    T -- Tidak --> V{Apakah task datang secara stream/queue?}
    V -- Ya --> W[✅ Worker Pool cocok]
    V -- Tidak --> Y{Apakah ada resource limit eksternal?}
    Y -- Ya --> Z[✅ Worker Pool cocok]
    Y -- Tidak --> X2[⚠️ Pertimbangkan dulu — mungkin tidak perlu]

Skenario yang Tepat #

I/O bound tasks adalah kandidat terbaik. HTTP calls ke external API, query ke database, baca/tulis disk, atau konsumsi message queue — semua ini melibatkan waktu tunggu yang lama, dan selama menunggu, worker bisa melayani job lain dari queue.

Stream atau queue-based input — Jika job datang terus-menerus dari Kafka, RabbitMQ, SQS, atau endpoint HTTP, Worker Pool adalah pasangan alami. Consumer membaca pesan, memasukkan ke job queue, dan worker memprosesnya dengan kecepatan yang terkontrol.

Resource dengan limit eksternal — Database connection pool, rate limit API pihak ketiga, dan semaphore eksplisit semuanya memiliki batas. Worker Pool memastikan kamu tidak melebihi batas tersebut dengan mengontrol berapa banyak task yang berjalan bersamaan.

Skenario yang Kurang Tepat #

Task yang sangat sedikit — Jika kamu hanya memproses 5 job sekali jalan, overhead setup Worker Pool tidak sepadan.

Task dengan dependensi kuat — Jika task B harus menunggu hasil task A sebelum bisa mulai, Worker Pool polos tidak cukup. Kamu butuh pipeline atau DAG scheduler.

Task membutuhkan prioritas kompleks — Standar channel di Go bersifat FIFO. Jika kamu butuh priority queue yang kompleks, Worker Pool dasar perlu dimodifikasi signifikan.

Implementasi Dasar di Golang #

Go memiliki goroutine dan channel yang membuat implementasi Worker Pool terasa sangat natural. Mari mulai dari implementasi paling sederhana, lalu kembangkan secara bertahap.

Struktur Dasar #

package main

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

// Job merepresentasikan pekerjaan yang perlu diproses
type Job struct {
    ID      int
    Payload string
}

// Result merepresentasikan hasil pemrosesan job
type Result struct {
    JobID  int
    Output string
    Err    error
}

// worker adalah fungsi yang dijalankan oleh setiap goroutine worker
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()

    for job := range jobs { // range akan keluar saat channel ditutup
        fmt.Printf("Worker %d mengambil job %d\n", id, job.ID)

        // simulasi pemrosesan
        time.Sleep(500 * time.Millisecond)
        output := fmt.Sprintf("Job %d selesai oleh Worker %d", job.ID, id)

        results <- Result{JobID: job.ID, Output: output}
    }
}

func main() {
    const workerCount = 3
    const jobCount = 10

    jobs := make(chan Job, jobCount)       // buffered channel untuk job queue
    results := make(chan Result, jobCount) // buffered channel untuk hasil

    var wg sync.WaitGroup

    // Mulai worker pool
    for i := 1; i <= workerCount; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Kirim semua job ke queue
    for j := 1; j <= jobCount; j++ {
        jobs <- Job{ID: j, Payload: fmt.Sprintf("data-%d", j)}
    }
    close(jobs) // sinyal bahwa tidak ada job baru

    // Tunggu semua worker selesai, lalu tutup results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Kumpulkan hasil
    for result := range results {
        fmt.Println(result.Output)
    }
}

Poin-poin kunci dari implementasi ini:

• jobs adalah job queue — channel yang menampung semua pekerjaan yang menunggu
• close(jobs) memberi sinyal ke semua worker bahwa tidak ada job baru; range jobs akan keluar secara otomatis
• sync.WaitGroup memastikan program menunggu sampai semua worker benar-benar selesai
• results channel memungkinkan worker mengembalikan hasil secara concurrent-safe

Alur Eksekusi #

sequenceDiagram
    participant M as Main / Producer
    participant Q as Job Queue (channel)
    participant W1 as Worker 1
    participant W2 as Worker 2
    participant W3 as Worker 3
    participant R as Results Channel

    M->>Q: Kirim Job 1–10
    M->>Q: close(jobs)

    par Worker berjalan bersamaan
        W1->>Q: Ambil Job 1
        W2->>Q: Ambil Job 2
        W3->>Q: Ambil Job 3
    end

    W1->>R: Result Job 1
    W2->>R: Result Job 2
    W3->>R: Result Job 3

    W1->>Q: Ambil Job 4
    W2->>Q: Ambil Job 5

    Note over Q: Queue kosong + closed
    W1-->>M: wg.Done()
    W2-->>M: wg.Done()
    W3-->>M: wg.Done()

    M->>R: close(results)
    M->>M: Kumpulkan semua result

Worker Pool dengan Context dan Graceful Shutdown #

Implementasi dasar di atas berfungsi, tapi production code membutuhkan lebih: kemampuan menghentikan worker secara bersih saat aplikasi menerima sinyal shutdown atau request di-cancel.

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

type Job struct {
    ID int
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

// worker yang context-aware
func worker(ctx context.Context, id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()

    // BENAR: selalu tangani panic di dalam worker
    defer func() {
        if r := recover(); r != nil {
            log.Printf("Worker %d: panic recovered: %v", id, r)
        }
    }()

    for {
        select {
        case <-ctx.Done():
            // Context dibatalkan — keluar dengan rapi
            log.Printf("Worker %d: shutdown karena context: %v", id, ctx.Err())
            return
        case job, ok := <-jobs:
            if !ok {
                // Channel ditutup — tidak ada job baru
                log.Printf("Worker %d: job channel ditutup, keluar", id)
                return
            }

            // Proses job
            result := processJob(ctx, id, job)
            results <- result
        }
    }
}

func processJob(ctx context.Context, workerID int, job Job) Result {
    // Simulasi pekerjaan dengan context awareness
    select {
    case <-ctx.Done():
        return Result{JobID: job.ID, Err: ctx.Err()}
    case <-time.After(200 * time.Millisecond):
        return Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("Job %d selesai oleh Worker %d", job.ID, workerID),
        }
    }
}

func main() {
    const workerCount = 3

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Tangkap sinyal OS untuk graceful shutdown
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)
    go func() {
        sig := <-sigChan
        log.Printf("Menerima sinyal: %v, memulai shutdown...", sig)
        cancel()
    }()

    jobs := make(chan Job, 50)
    results := make(chan Result, 50)
    var wg sync.WaitGroup

    // Start worker pool
    for i := 1; i <= workerCount; i++ {
        wg.Add(1)
        go worker(ctx, i, jobs, results, &wg)
    }

    // Producer goroutine
    go func() {
        defer close(jobs)
        for j := 1; j <= 20; j++ {
            select {
            case <-ctx.Done():
                log.Println("Producer: context dibatalkan, berhenti mengirim job")
                return
            case jobs <- Job{ID: j}:
                log.Printf("Job %d dikirim ke queue", j)
            }
        }
    }()

    // Tunggu worker selesai lalu tutup results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Kumpulkan hasil
    for result := range results {
        if result.Err != nil {
            log.Printf("Job %d error: %v", result.JobID, result.Err)
            continue
        }
        fmt.Println(result.Output)
    }

    log.Println("Semua worker selesai.")
}

Diagram shutdown yang bersih:

stateDiagram-v2
    [*] --> Running: Worker pool dimulai
    Running --> Draining: close(jobs) atau ctx.Done()
    Draining --> Finishing: Semua job di queue diproses
    Finishing --> Done: wg.Wait() selesai
    Done --> [*]

    Running --> Interrupted: SIGTERM / SIGINT
    Interrupted --> Draining: cancel() dipanggil

Worker Pool vs Thread Pool #

Kedua istilah sering dipertukarkan, tapi ada perbedaan penting — terutama saat berbicara tentang implementasi di Go.

Perbandingan Konseptual #

Perbedaan Filosofi #

Thread Pool di Java atau C# biasanya dikelola oleh runtime atau framework. Kamu menyerahkan tugas ke pool dan runtime yang menentukan thread mana yang menjalankannya. Ini abstraksi yang tinggi, tapi kamu kehilangan kontrol granular.

Worker Pool di Go menggunakan goroutine yang sangat ringan dan channel sebagai primitif komunikasi. Kamu membangun sendiri mekanismenya, tapi justru itulah yang memberikan kontrol penuh — ukuran pool, buffer queue, behavior shutdown, semua ada di tanganmu.

flowchart TD
    subgraph Java["Java — Thread Pool"]
        J1[Task 1] --> ES[ExecutorService]
        J2[Task 2] --> ES
        ES --> JT1[OS Thread 1]
        ES --> JT2[OS Thread 2]
        ES --> JT3[OS Thread N]
    end

    subgraph Go["Go — Worker Pool"]
        G1[Job 1] --> CH[(Channel / Queue)]
        G2[Job 2] --> CH
        CH --> GR1[Goroutine 1]
        CH --> GR2[Goroutine 2]
        CH --> GR3[Goroutine N]
    end

Secara prinsip: Worker Pool adalah pattern umum, Thread Pool adalah salah satu implementasinya. Di Go, kamu mengimplementasikan Worker Pool menggunakan goroutine (bukan OS thread), yang jauh lebih skalabel.

Variasi Lanjutan #

Worker Pool dasar bisa dikembangkan menjadi berbagai variasi untuk kebutuhan yang lebih kompleks.

Dynamic Worker Pool #

Worker Pool standar memiliki jumlah worker yang tetap. Dynamic Worker Pool menyesuaikan jumlah worker berdasarkan panjang antrian:

type DynamicPool struct {
    minWorkers int
    maxWorkers int
    jobs       chan Job
    mu         sync.Mutex
    active     int
    ctx        context.Context
    cancel     context.CancelFunc
}

func (p *DynamicPool) scale() {
    p.mu.Lock()
    defer p.mu.Unlock()

    queueLen := len(p.jobs)

    // Tambah worker jika queue menumpuk dan belum mencapai batas
    if queueLen > p.active*2 && p.active < p.maxWorkers {
        p.active++
        go p.runWorker(p.active)
    }
}

Rate-Limited Worker Pool #

Berguna saat memproses job yang memanggil external API dengan rate limit:

import "golang.org/x/time/rate"

func rateLimitedWorker(
    ctx context.Context,
    id int,
    limiter *rate.Limiter,
    jobs <-chan Job,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    for job := range jobs {
        // Tunggu token dari rate limiter sebelum memproses
        if err := limiter.Wait(ctx); err != nil {
            log.Printf("Worker %d: rate limiter error: %v", id, err)
            return
        }
        processJob(ctx, id, job)
    }
}

// Inisialisasi: 10 request per detik, burst maksimal 20
limiter := rate.NewLimiter(rate.Limit(10), 20)

Pipeline Worker Pool (Multi-Stage) #

Beberapa Worker Pool yang terhubung secara serial — output dari pool pertama menjadi input pool berikutnya:

flowchart LR
    I[Input] --> Q1[(Queue 1)]
    Q1 --> S1[Stage 1\nWorker Pool]
    S1 --> Q2[(Queue 2)]
    Q2 --> S2[Stage 2\nWorker Pool]
    S2 --> Q3[(Queue 3)]
    Q3 --> S3[Stage 3\nWorker Pool]
    S3 --> O[Output]

// stage1 memproses data mentah
func stage1(ctx context.Context, input <-chan RawData) <-chan ParsedData {
    output := make(chan ParsedData, 100)
    go func() {
        defer close(output)
        pool := newWorkerPool(ctx, 5, input, output, parseData)
        pool.Wait()
    }()
    return output
}

// stage2 melakukan transformasi pada parsed data
func stage2(ctx context.Context, input <-chan ParsedData) <-chan FinalResult {
    output := make(chan FinalResult, 100)
    go func() {
        defer close(output)
        pool := newWorkerPool(ctx, 3, input, output, transformData)
        pool.Wait()
    }()
    return output
}

Retry dan Dead Letter Queue #

flowchart TD
    J[Job Masuk] --> Q[(Job Queue)]
    Q --> W[Worker]
    W -- Sukses --> R[Result]
    W -- Gagal, retry < max --> RQ[(Retry Queue)]
    RQ --> W
    W -- Gagal, retry = max --> DLQ[(Dead Letter Queue)]
    DLQ --> Alert[Alert / Manual Review]

type RetryableJob struct {
    Job
    RetryCount int
    MaxRetry   int
}

func workerWithRetry(jobs <-chan RetryableJob, dlq chan<- RetryableJob) {
    for job := range jobs {
        err := processJob(job.Job)
        if err != nil {
            if job.RetryCount < job.MaxRetry {
                // Kembalikan ke queue dengan retry count +1
                job.RetryCount++
                jobs <- job // HATI-HATI: ini bisa deadlock jika channel tidak buffered
            } else {
                // Kirim ke dead letter queue
                dlq <- job
            }
        }
    }
}

Best Practice #

Memahami implementasi Worker Pool tidak cukup — yang membedakan sistem yang baik dari yang buruk adalah bagaimana kamu menerapkan pattern ini secara benar.

1. Tentukan Jumlah Worker Berdasarkan Jenis Task #

// ANTI-PATTERN: angka ajaib tanpa dasar
workerCount := 100 // dari mana angka ini?

// BENAR: tentukan berdasarkan karakteristik task
numCPU := runtime.NumCPU()

// CPU-bound task: sesuaikan dengan jumlah CPU
cpuBoundWorkers := numCPU

// I/O bound task: bisa lebih banyak karena worker sering menunggu
// Mulai dengan 2x-4x CPU, lalu benchmark
ioBoundWorkers := numCPU * 3

Aturan praktis: CPU-bound → runtime.NumCPU(), I/O-bound → mulai dari 2-4x NumCPU() lalu benchmark dengan load nyata.

2. Pilih Buffer Channel yang Tepat #

// ANTI-PATTERN: unbuffered channel untuk job queue tinggi volume
jobs := make(chan Job) // producer akan block setiap kali worker belum siap

// ANTI-PATTERN: buffer terlalu besar yang menyembunyikan bottleneck
jobs := make(chan Job, 1_000_000) // masalah performa tersembunyi di balik buffer besar

// BENAR: buffer yang proporsional — cukup untuk menyerap burst singkat
jobs := make(chan Job, workerCount*2) // atau sesuaikan dengan throughput yang diharapkan

Monitor panjang queue secara rutin. Jika queue selalu penuh, itu sinyal bahwa consumer (worker) tidak mampu mengimbangi producer.

3. Selalu Tangani Panic di Worker #

Satu worker yang panic tidak boleh mematikan seluruh pool:

// ANTI-PATTERN: worker tanpa panic recovery
func naiveWorker(jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        processJob(job) // jika ini panic, goroutine mati dan wg.Done() tidak dipanggil
    }
}

// BENAR: worker dengan panic recovery
func safeWorker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    defer func() {
        if r := recover(); r != nil {
            log.Printf("Worker %d: recovered from panic: %v\nStack: %s",
                id, r, debug.Stack())
            // Restart worker jika diperlukan
        }
    }()

    for job := range jobs {
        processJob(job)
    }
}

4. Jangan Kirim ke Channel yang Sudah Ditutup #

// ANTI-PATTERN: menutup channel dari sisi producer saat ada multiple producer
go func() { // Producer 1
    jobs <- Job{ID: 1}
    close(jobs) // Producer 2 mungkin masih mau kirim!
}()

go func() { // Producer 2
    jobs <- Job{ID: 2} // PANIC: send on closed channel
}()

// BENAR: gunakan sync.Once atau WaitGroup untuk koordinasi multiple producer
var once sync.Once
closeJobs := func() { once.Do(func() { close(jobs) }) }

// atau gunakan sync.WaitGroup untuk menunggu semua producer selesai
var producerWg sync.WaitGroup
go func() {
    producerWg.Wait()
    close(jobs)
}()

5. Implementasikan Backpressure #

Jika producer jauh lebih cepat dari consumer, kamu butuh mekanisme backpressure:

// ANTI-PATTERN: producer terus kirim tanpa peduli queue penuh
for _, item := range bigDataset {
    jobs <- Job{Data: item} // block tak terbatas jika queue penuh
}

// BENAR: producer dengan timeout dan backpressure
for _, item := range bigDataset {
    select {
    case jobs <- Job{Data: item}:
        // berhasil dikirim
    case <-time.After(5 * time.Second):
        // queue masih penuh setelah 5 detik — log warning
        log.Println("Backpressure: job queue penuh selama 5 detik")
        metrics.Increment("worker_pool.backpressure")
    case <-ctx.Done():
        return
    }
}

6. Observability Adalah Keharusan #

Worker Pool yang bermasalah sering menjadi silent bottleneck — sistem berjalan, tapi lambat secara misterius:

type PoolMetrics struct {
    JobsEnqueued  atomic.Int64
    JobsProcessed atomic.Int64
    JobsFailed    atomic.Int64
    QueueLength   func() int
}

func instrumentedWorker(
    id int,
    jobs <-chan Job,
    results chan<- Result,
    metrics *PoolMetrics,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    for job := range jobs {
        start := time.Now()

        result := processJob(job)

        duration := time.Since(start)
        metrics.JobsProcessed.Add(1)

        if result.Err != nil {
            metrics.JobsFailed.Add(1)
        }

        // Kirim ke observability system (Prometheus, Datadog, dll)
        histogram.Observe("worker_job_duration_seconds", duration.Seconds(),
            map[string]string{"worker_id": fmt.Sprint(id)})

        results <- result
    }
}

Metrik yang wajib dipantau:

Anti-Pattern yang Harus Dihindari #

Berikut anti-pattern paling umum yang sering muncul saat mengimplementasikan Worker Pool:

// ✗ Tidak menutup job channel — worker hang selamanya
func badMain() {
    jobs := make(chan Job)
    var wg sync.WaitGroup
    wg.Add(1)
    go worker(jobs, &wg)
    jobs <- Job{ID: 1}
    // lupa: close(jobs) -- worker akan menunggu job baru selamanya
    wg.Wait() // deadlock!
}
// ✓ Selalu close(jobs) setelah semua job dikirim

// ✗ WaitGroup tidak sinkron dengan jumlah worker
wg.Add(1) // hanya tambah 1
for i := 0; i < 5; i++ {
    go worker(jobs, &wg) // tapi ada 5 goroutine yang panggil wg.Done()
}
// ✓ wg.Add(workerCount) sebelum loop, atau wg.Add(1) di dalam loop sebelum go

// ✗ Membaca dari results setelah results ditutup dua kali
close(results)
close(results) // PANIC: close of closed channel
// ✓ Gunakan sync.Once untuk menutup channel

// ✗ Worker yang tidak context-aware — tidak bisa dihentikan
func badWorker(jobs <-chan Job) {
    for job := range jobs {
        http.Get("https://api.example.com/process") // bisa block lama tanpa timeout
    }
}
// ✓ Selalu gunakan context dengan timeout untuk operasi eksternal

// ✗ Shared mutable state tanpa proteksi di antara worker
var counter int
func worker(jobs <-chan Job) {
    for range jobs {
        counter++ // race condition!
    }
}
// ✓ Gunakan atomic, mutex, atau kirim hasil lewat channel
var counter atomic.Int64
counter.Add(1)

Checklist Implementasi Worker Pool #

SETUP:
  □ Jumlah worker ditentukan berdasarkan jenis task (CPU/IO bound)
  □ Buffer channel dipilih berdasarkan throughput yang diharapkan
  □ WaitGroup.Add() dipanggil sebelum goroutine dimulai

SHUTDOWN:
  □ close(jobs) dipanggil setelah semua job dikirim
  □ Hanya satu goroutine yang menutup channel (cegah double close)
  □ context.Cancel() diintegrasikan dengan sinyal OS (SIGTERM/SIGINT)
  □ wg.Wait() dipanggil sebelum menutup results channel

SAFETY:
  □ recover() ada di setiap worker goroutine
  □ Semua operasi eksternal menggunakan context dengan timeout
  □ Shared state dilindungi atomic atau mutex (atau dihindari sama sekali)
  □ Producer menggunakan select dengan ctx.Done() untuk mencegah block abadi

OBSERVABILITY:
  □ Queue length dipantau secara periodik
  □ Throughput dan latency diukur per job
  □ Error rate dicatat dan di-alert
  □ Graceful shutdown di-log dengan jelas

Ringkasan #

• Worker Pool = kontrol concurrency — jumlah worker yang terbatas melindungi resource sistem dari lonjakan tak terkendali.
• Tiga komponen inti: Job Queue (channel), Worker (goroutine), dan Dispatcher (producer yang mengirim job).
• Paling efektif untuk I/O bound tasks — HTTP calls, DB queries, message queue consumer — di mana worker sering menunggu dan bisa melayani job lain.
• Buffered channel untuk job queue membantu menyerap burst singkat, tapi buffer terlalu besar bisa menyembunyikan bottleneck.
• Selalu implement graceful shutdown dengan context.Context dan tangkap sinyal OS (SIGTERM, SIGINT) agar worker berhenti dengan rapi.
• Panic recovery wajib ada di setiap worker — satu panic tidak boleh mematikan seluruh pool.
• Backpressure bukan opsional — gunakan select dengan timeout di sisi producer agar queue yang penuh tidak menyebabkan block tak terbatas.
• Worker Pool vs Thread Pool — keduanya menerapkan konsep yang sama, tapi di Go implementasinya menggunakan goroutine (jauh lebih ringan dari OS thread) dan channel (lebih aman dari shared memory).
• Observability adalah kewajiban — pantau panjang queue, throughput, latency, dan error rate; tanpa metrik, Worker Pool bisa menjadi silent bottleneck.
• Variasi lanjutan — Dynamic Pool, Rate-Limited Pool, Pipeline Pool, dan Retry + Dead Letter Queue — semuanya dibangun di atas fondasi yang sama.

← Sebelumnya: Thread Pool   Berikutnya: Producer-Consumer →