Bagian 1: Desain Worker Pool

Implementasi concurrency yang ditulis langsung tanpa rancangan biasanya berakhir dengan channel yang tidak jelas siapa menutup, goroutine yang tidak jelas kapan keluar, dan pengujian yang tidak jelas apa yang dijamin. Pada bagian ini kita menulis tiga keputusan terpenting lebih dulu dalam bentuk kode: seperti apa satu job, seperti apa hasilnya, dan seperti apa wadah yang menghubungkan keduanya. Seluruh implementasi pada bagian berikutnya hanya mengikuti kontrak yang sudah terpatri di sini.

Tipe Job dan Result #

Worker pool berputar di sekitar dua tipe kecil di worker.go:

go
// Job adalah satu unit pekerjaan yang dikirim ke pool.
// ID menjaga identitas job sehingga hasil bisa diurutkan kembali,
// Payload membawa data yang diproses fungsi kerja.
type Job struct {
	ID      int
	Payload string
}

// Result adalah hasil pemrosesan satu Job. Err tidak kosong ketika
// fungsi kerja gagal memproses Payload.
type Result struct {
	JobID  int
	Output string
	Err    error
}

Mari kita telaah keputusan di balik tiap fieldnya:

Job.ID (baris 12): tiga worker menyelesaikan sepuluh job dalam urutan apa pun; ID memungkinkan hasil dikumpulkan lalu diurutkan kembali sehingga output program tetap deterministik.
Job.Payload (baris 13): data pekerjaan dipisah dari identitasnya. Payload bisa berubah jenis (struct lebih besar, byte, URL) tanpa menyentuh mekanisme pool.
Result.JobID (baris 19): hasil membawa asal-usulnya sendiri. Pembaca hasil tidak perlu menebak job mana yang menghasilkan apa.
Result.Err (baris 21): kegagalan satu job bukan kegagalan pool. Galat dibawa sebagai data lewat Err, bukan dilempar sebagai panic, sehingga worker lain tetap memproses job mereka.

Pemisahan Job dan Result juga menandai batas tanggung jawab: pengirim hanya tahu Job, pembaca hanya tahu Result. Dua sisi program bisa berkembang terpisah selama kedua tipe ini stabil.

Struct Pool dan Konstruktornya #

Wadah penghubung keduanya adalah Pool di pool.go:

go
// Pool membatasi jumlah worker yang memproses job secara bersamaan.
// Job masuk lewat channel jobs, keluar sebagai Result lewat channel
// results; keduanya tanpa buffer sehingga pengirim, worker, dan pembaca
// hasil bertemu langsung tanpa antrean tersembunyi.
type Pool struct {
	jumlahWorker int
	fn           func(Job) (string, error)
	jobs         chan Job
	results      chan Result
	ctx          context.Context
	cancel       context.CancelFunc
	wg           sync.WaitGroup
	sekali       sync.Once
}

// New membuat pool ber-jumlahWorker worker yang memproses job dengan
// fungsi fn. Pool belum berjalan sampai Start dipanggil.
func New(jumlahWorker int, fn func(Job) (string, error)) *Pool {
	return &Pool{
		jumlahWorker: jumlahWorker,
		fn:           fn,
		jobs:         make(chan Job),
		results:      make(chan Result),
	}
}

Enam field struct ini adalah seluruh state yang dibutuhkan pool — tidak ada mutex, tidak ada map status, tidak ada slice hasil:

fn (baris 14): fungsi kerja disuntikkan lewat New, bukan dihardcode di dalam worker. Pool menjadi bebas memproses apa pun: kapitalisasi teks di demo, panggilan API di produksi, atau fungsi palsu di pengujian.
jobs dan results (baris 15-16): dua channel tanpa buffer. Pengirim merasakan langsung kapan semua worker sibuk — ini backpressure, pengaman alami dari beban berlebih.
ctx dan cancel (baris 17-18): context hasil turunan Start; seluruh worker berbagi satu tombol darurat yang sama.
wg (baris 19): sync.WaitGroup menghitung worker yang masih hidup; pengawas pool menunggu angka ini nol sebelum menutup channel hasil.
sekali (baris 20): sync.Once menjaga Shutdown tidak menutup channel job dua kali — penutupan ganda akan panic.
New tanpa channel berjalan (baris 25-32): konstruktor hanya menyiapkan struktur; goroutine baru lahir di Start. Pemisahan ini membuat pool bisa dibuat dulu, dikonfigurasi, lalu dinyalakan pada saat yang tepat.

Keputusan Desain yang Menempel #

Rangkaian keputusan di atas membentuk kontrak yang ringkas:

Keputusan	Konsekuensi
Channel tanpa buffer	Pengirim diblokir saat semua worker sibuk; beban terkendali
Galat sebagai data (`Result.Err`)	Satu job gagal tidak menghentikan pool
Fungsi kerja disuntikkan	Pool bisa diuji dengan fungsi palsu tanpa pekerjaan nyata
ID pada job dan hasil	Output dapat diurutkan menjadi deterministik
`sync.Once` di `Shutdown`	Penutupan channel aman dipanggil berulang

Rancangan yang Terkompilasi #

Kontrak terpatri di tiga berkas — worker.go, pool.go, dan main.go — dalam satu module:

go
module example.com/workerpool

go 1.27

Sebelum satu goroutine pun dinyalakan, pastikan rancangan ini kompilasi bersih:

bash
go vet ./...; echo "exit: $?"

Output yang Dihasilkan:

text

exit: 0

go vet memeriksa pola mencurigakan — penutupan channel ganda, format Printf salah, penyalinan mutex — dan keluar dengan kode 0 berarti rancangan kita lolos pemeriksaan statis itu. Mulai sekarang, setiap perubahan implementasi cukup dijaga oleh kontrak yang sudah tercetak di atas.

Rujukan Materi #

Worker Pool — versi ringkas pola ini sebagai pembanding rancangan.
Struct dan Method — dasar perancangan tipe yang membawa data dan perilaku.
Error Interface — alasan galat dianggap nilai biasa di Go.
Dokumentasi sync — referensi resmi WaitGroup dan Once.

Kontrak sudah tertulis: Job masuk, Result keluar, dan Pool menjaga keduanya. Bagian 2 menghidupkannya dengan goroutine worker yang mengonsumsi channel job.

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

go
package main

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"strings"
)

const (
	jumlahWorker = 3
	jumlahJob    = 10
)

// prosesJob adalah pekerjaan yang dijalankan setiap worker: mengubah
// payload menjadi huruf besar. Pada kasus nyata fungsi ini bisa berisi
// kalkulasi berat, pemanggilan API, atau pembacaan berkas.
func prosesJob(job Job) (string, error) {
	return strings.ToUpper(job.Payload), nil
}

func main() {
	fmt.Println("=== Demo Worker Pool ===")
	fmt.Printf("pool: %d worker, %d job\n", jumlahWorker, jumlahJob)

	sebelum := runtime.NumGoroutine()

	pool := New(jumlahWorker, prosesJob)
	pool.Start(context.Background())

	// Pengiriman job berjalan di goroutine terpisah karena kedua channel
	// pool tanpa buffer: pengirim dan pembaca hasil harus aktif bersamaan,
	// jika tidak keduanya saling menunggu selamanya (deadlock).
	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= jumlahJob; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	// Kumpulkan seluruh hasil; perulangan berhenti saat pengawas pool
	// menutup channel hasil.
	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	// Urutan kedatangan hasil dari beberapa goroutine tidak menentu,
	// jadi hasil dikumpulkan dulu lalu diurutkan sebelum dicetak agar
	// output selalu sama setiap eksekusi.
	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	fmt.Println("hasil (terurut berdasarkan job):")
	for _, h := range hasil {
		fmt.Printf("job %02d → %s\n", h.JobID, h.Output)
	}

	// Keluarnya goroutine tidak bisa diamati lewat channel mana pun,
	// jadi beri penjadwal kesempatan menjalankan epilogue goroutine
	// yang sudah menyelesaikan tugasnya sebelum menghitung ulang.
	tungguGoroutine(sebelum)

	sesudah := runtime.NumGoroutine()
	fmt.Printf("goroutine sebelum=%d, sesudah=%d\n", sebelum, sesudah)
}

// tungguGoroutine menunggu jumlah goroutine kembali ke nilai dasar
// dengan batas putaran tetap. Ini sinkronisasi eksplisit untuk demo;
// pengujian memakai fungsi yang sama.
func tungguGoroutine(dasar int) {
	for i := 0; i < 1000 && runtime.NumGoroutine() > dasar; i++ {
		runtime.Gosched()
	}
}

go
package main

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"strings"
	"testing"
)

func fungsiUji(job Job) (string, error) {
	return strings.ToUpper(job.Payload), nil
}

func TestPoolHasilLengkapDanTerurut(t *testing.T) {
	pool := New(3, fungsiUji)
	pool.Start(context.Background())

	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= 10; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	if len(hasil) != 10 {
		t.Fatalf("jumlah hasil = %d, ingin 10", len(hasil))
	}
	for i, h := range hasil {
		want := fmt.Sprintf("TUGAS-%02d", i+1)
		if h.JobID != i+1 || h.Output != want || h.Err != nil {
			t.Errorf("hasil[%d] = %+v, ingin JobID %d Output %q", i, h, i+1, want)
		}
	}
}

func TestPoolShutdownBersihTanpaBocor(t *testing.T) {
	dasar := runtime.NumGoroutine()

	pool := New(3, fungsiUji)
	pool.Start(context.Background())

	go func() {
		for i := 1; i <= 6; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	jumlah := 0
	for range pool.Results() {
		jumlah++
	}

	if jumlah != 6 {
		t.Fatalf("jumlah hasil = %d, ingin 6", jumlah)
	}

	// Shutdown dipanggil kedua kali untuk memastikan pool tetap aman.
	pool.Shutdown()

	tungguGoroutine(dasar)
	if sesudah := runtime.NumGoroutine(); sesudah > dasar {
		t.Errorf("goroutine tersisa: sebelum=%d sesudah=%d", dasar, sesudah)
	}
}

func TestPoolPembatalanContext(t *testing.T) {
	dasar := runtime.NumGoroutine()

	ctx, cancel := context.WithCancel(context.Background())
	pool := New(3, fungsiUji)
	pool.Start(ctx)

	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= 6; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		// Batalkan di tengah jalan: sebagian job mungkin sudah diproses,
		// sebagian dibuang Submit. Yang diuji adalah pool berhenti tanpa
		// menggantung dan tanpa goroutine tersisa.
		cancel()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	if len(hasil) > 6 {
		t.Errorf("jumlah hasil = %d, tidak boleh melebihi 6", len(hasil))
	}

	tungguGoroutine(dasar)
	if sesudah := runtime.NumGoroutine(); sesudah > dasar {
		t.Errorf("goroutine tersisa setelah pembatalan: sebelum=%d sesudah=%d", dasar, sesudah)
	}
}

func TestPoolJobGagalDilaporkan(t *testing.T) {
	fn := func(job Job) (string, error) {
		if job.Payload == "" {
			return "", fmt.Errorf("job %d: payload kosong", job.ID)
		}
		return strings.ToUpper(job.Payload), nil
	}

	pool := New(2, fn)
	pool.Start(context.Background())

	go func() {
		pool.Submit(Job{ID: 1, Payload: "tugas-01"})
		pool.Submit(Job{ID: 2, Payload: ""})
		pool.Shutdown()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}

	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	if len(hasil) != 2 {
		t.Fatalf("jumlah hasil = %d, ingin 2", len(hasil))
	}
	if hasil[0].Err != nil || hasil[0].Output != "TUGAS-01" {
		t.Errorf("hasil[0] = %+v, ingin sukses TUGAS-01", hasil[0])
	}
	if hasil[1].Err == nil {
		t.Errorf("hasil[1].Err = nil, ingin galat payload kosong")
	}
}

go
package main

import (
	"context"
	"sync"
)

// Pool membatasi jumlah worker yang memproses job secara bersamaan.
// Job masuk lewat channel jobs, keluar sebagai Result lewat channel
// results; keduanya tanpa buffer sehingga pengirim, worker, dan pembaca
// hasil bertemu langsung tanpa antrean tersembunyi.
type Pool struct {
	jumlahWorker int
	fn           func(Job) (string, error)
	jobs         chan Job
	results      chan Result
	ctx          context.Context
	cancel       context.CancelFunc
	wg           sync.WaitGroup
	sekali       sync.Once
}

// New membuat pool ber-jumlahWorker worker yang memproses job dengan
// fungsi fn. Pool belum berjalan sampai Start dipanggil.
func New(jumlahWorker int, fn func(Job) (string, error)) *Pool {
	return &Pool{
		jumlahWorker: jumlahWorker,
		fn:           fn,
		jobs:         make(chan Job),
		results:      make(chan Result),
	}
}

// Start menjalankan seluruh worker plus satu goroutine pengawas.
// Context yang dibatalkan menghentikan semua worker di tengah pekerjaan;
// Shutdown tetap perlu dipanggil untuk jalur berhenti yang normal.
func (p *Pool) Start(ctx context.Context) {
	p.ctx, p.cancel = context.WithCancel(ctx)
	for i := 0; i < p.jumlahWorker; i++ {
		p.wg.Add(1)
		go worker(p.ctx, p.jobs, p.results, p.fn, &p.wg)
	}
	// Pengawas menutup channel hasil setelah semua worker keluar.
	// Pembaca hasil berhenti bereputar tepat pada saat itu — tanpa
	// pengawas, pembaca tidak pernah tahu pool sudah selesai.
	go p.awasi()
}

// awasi menunggu worker habis lalu menutup results dan melepas context.
// Goroutine inilah yang mencegah bocor: selesai → channel ditutup.
func (p *Pool) awasi() {
	p.wg.Wait()
	close(p.results)
	p.cancel()
}

// Submit mengirim satu job ke worker yang tersedia dan memblokir saat
// semua worker sedang sibuk (backpressure). Job dibuang begitu context
// pool sudah dibatalkan, jadi pemanggil harus berhenti mengirim setelah
// pembatalan terjadi.
func (p *Pool) Submit(job Job) {
	select {
	case p.jobs <- job:
	case <-p.ctx.Done():
		// Pool sudah berhenti; job tidak dikirim.
	}
}

// Results mengembalikan channel hasil. Channel ini ditutup otomatis
// setelah seluruh worker keluar, sehingga `for range` berakhir pasti.
func (p *Pool) Results() <-chan Result {
	return p.results
}

// Shutdown menutup channel job. Worker menghabiskan sisa job yang sudah
// masuk, keluar, lalu pengawas menutup channel hasil. Shutdown aman
// dipanggil berapa pun kali — hanya pemanggil pertama yang menutup.
func (p *Pool) Shutdown() {
	p.sekali.Do(func() {
		close(p.jobs)
	})
}

go
package main

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"
)

func TestWorkerMemprosesJob(t *testing.T) {
	jobs := make(chan Job, 1)
	results := make(chan Result, 1)
	var wg sync.WaitGroup

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 7, Payload: "tugas-07"}
	close(jobs)

	res := <-results
	wg.Wait()

	if res.JobID != 7 {
		t.Errorf("JobID hasil = %d, ingin 7", res.JobID)
	}
	if res.Output != "TUGAS-07" {
		t.Errorf("Output hasil = %q, ingin TUGAS-07", res.Output)
	}
	if res.Err != nil {
		t.Errorf("Err hasil = %v, ingin nil", res.Err)
	}
}

func TestWorkerJobGagal(t *testing.T) {
	jobs := make(chan Job, 1)
	results := make(chan Result, 1)
	var wg sync.WaitGroup

	galatUji := errors.New("payload kosong")
	fn := func(job Job) (string, error) {
		return "", galatUji
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: ""}
	close(jobs)

	res := <-results
	wg.Wait()

	if res.Err == nil {
		t.Fatal("Err hasil = nil, ingin galat")
	}
	if !errors.Is(res.Err, galatUji) {
		t.Errorf("Err hasil = %v, ingin galat uji", res.Err)
	}
	if res.Output != "" {
		t.Errorf("Output hasil = %q, ingin string kosong saat gagal", res.Output)
	}
}

func TestWorkerBerhentiSaatContextDibatalkan(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	// Batalkan sebelum worker sempat mengambil job apa pun.
	cancel()

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

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(ctx, jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: "tugas-01"}
	close(jobs)
	wg.Wait()

	select {
	case res := <-results:
		t.Errorf("worker memproses job setelah pembatalan: %+v", res)
	default:
		// Tidak ada hasil: worker keluar tanpa memproses apa pun.
	}
}

func TestWorkerChannelTertutupMenghentikanWorker(t *testing.T) {
	jobs := make(chan Job, 2)
	results := make(chan Result, 2)
	var wg sync.WaitGroup

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: "satu"}
	jobs <- Job{ID: 2, Payload: "dua"}
	close(jobs)
	wg.Wait()

	if len(results) != 2 {
		t.Fatalf("jumlah hasil = %d, ingin 2", len(results))
	}
}

go
package main

import (
	"context"
	"sync"
)

// Job adalah satu unit pekerjaan yang dikirim ke pool.
// ID menjaga identitas job sehingga hasil bisa diurutkan kembali,
// Payload membawa data yang diproses fungsi kerja.
type Job struct {
	ID      int
	Payload string
}

// Result adalah hasil pemrosesan satu Job. Err tidak kosong ketika
// fungsi kerja gagal memproses Payload.
type Result struct {
	JobID  int
	Output string
	Err    error
}

// worker mengonsumsi Job dari channel jobs sampai channel ditutup atau
// context dibatalkan, lalu menuliskan Result ke channel results.
// wg.Done via defer menjamin pool tahu worker ini sudah keluar.
func worker(ctx context.Context, jobs <-chan Job, results chan<- Result, fn func(Job) (string, error), wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		// Periksa pembatalan lebih dulu agar worker berhenti seketika
		// ketika context sudah dibatalkan sebelum job berikutnya diambil.
		if ctx.Err() != nil {
			return
		}
		select {
		case <-ctx.Done():
			return
		case job, ok := <-jobs:
			if !ok {
				return
			}
			output, err := fn(job)
			// Pengiriman hasil juga harus siap kalah dari pembatalan;
			// tanpa case ini worker bisa menggantung selamanya menunggu
			// pembaca hasil yang sudah tidak ada.
			select {
			case <-ctx.Done():
				return
			case results <- Result{JobID: job.ID, Output: output, Err: err}:
			}
		}
	}
}

Bagian 1: Desain Worker Pool

Tipe Job dan Result #

Struct Pool dan Konstruktornya #

Keputusan Desain yang Menempel #

Rancangan yang Terkompilasi #

Rujukan Materi #

Berkas Sumber

main.go

pool_test.go

pool.go

worker_test.go

worker.go