Bagian 2: Implementasi Goroutine

Kontrak dari Bagian 1 masih berupa struct diam. Bagian ini menghidupkannya: satu fungsi worker menjadi goroutine yang mengonsumsi channel job, Start menyalakan beberapa salinan sekaligus, dan Submit menyebarkan pekerjaan ke worker yang tersedia. Di ujung bagian, demo sepuluh job pada tiga worker membuktikan distribusi bekerja dan hasil tetap terurut.

Goroutine Worker #

Seluruh pekerjaan pool berpusat pada satu fungsi di worker.go:

go
// 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}:
			}
		}
	}
}

Mari kita bedah loop worker baris demi baris di worker.go:

Baris 28 (defer wg.Done()): tanda keluar worker terpasang sejak detik pertama. Bagaimanapun worker berhenti — channel tertutup, context batal, atau fn selesai — pengawas pool selalu tahu worker ini sudah mati.
Baris 32-34 (if ctx.Err() != nil): pemeriksaan non-blokir sebelum mengambil job. Tanpa pemeriksaan ini, select bisa memilih channel job meski pembatalan sudah terjadi, dan worker memproses satu job ekstra setelah diperintahkan berhenti.
Baris 35-37 (case <-ctx.Done()): jalur darurat ketika worker sedang tertidur menunggu job. Pembatalan context membangunkannya dan worker keluar.
Baris 38-41 (job, ok := <-jobs): menerima job; ok bernilai false saat channel ditutup Shutdown — itulah sinyal berhenti normal. Worker menghabiskan job yang tersisa sebelum keluar, bukan membuangnya.
Baris 42 (fn(job)): pekerjaan sungguhan. Fungsi injeksi dari Bagian 1 dipanggil di sini; worker tidak peduli apa isinya.
Baris 46-50 (select kedua): pengiriman hasil dipertaruhkan melawan pembatalan. Bila pembaca hasil sudah pergi karena pool dibatalkan, pengiriman blokiran akan menggantung selamanya — kasus goroutine leak klasik.

Perhatikan parameter channel bertipe arah: jobs <-chan Job (hanya menerima) dan results chan<- Result (hanya mengirim). Kompilator ikut menjaga worker tidak pernah mengirim job atau menerima hasil.

Menyalakan Pool #

Start di pool.go melahirkan goroutine-goroutine itu:

go
// 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()
}

Tiga hal terjadi berurutan: context turunan dibuat (baris 38) sehingga pool bisa membatalkan dirinya sendiri tanpa menunggu induk, penghitung wg dinaikkan sebelum setiap go (baris 40-41) — urutan yang wajib, sebab Add setelah Wait berjalan menghasilkan perlombaan — dan terakhir pengawas awasi dinyalakan (baris 46). Jumlah worker ditentukan sekali di New; tidak ada method untuk menambah atau mengurangi worker saat berjalan karena ukuran tetap membuat penutupan jauh lebih mudah dibuktikan.

Mengirim Job dengan Backpressure #

go
// 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.
	}
}

Channel jobs tanpa buffer membuat Submit menunggu sampai ada worker yang benar-benar siap menerima. Saat tiga worker sibuk, pengirim keempat berhenti sebentar — itulah backpressure yang kita rancang di Bagian 1, bekerja tanpa satu baris kode pengecekan tambahan. Case kedua memberi Submit pintu keluar saat pool sudah dibatalkan: job dibuang dan pengirim harus berhenti.

Demo Distribusi Sepuluh Job #

main.go merakit seluruh alur itu. Bagian intinya:

go
	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
	})

Goroutine kiri mengirim sepuluh job lalu memanggil Shutdown; goroutine kanan (main) membaca hasil sampai channel ditutup. Jalankan dari folder proyek:

bash

go run .

Output yang Dihasilkan:

text
=== Demo Worker Pool ===
pool: 3 worker, 10 job
hasil (terurut berdasarkan job):
job 01 → TUGAS-01
job 02 → TUGAS-02
job 03 → TUGAS-03
job 04 → TUGAS-04
job 05 → TUGAS-05
job 06 → TUGAS-06
job 07 → TUGAS-07
job 08 → TUGAS-08
job 09 → TUGAS-09
job 10 → TUGAS-10
goroutine sebelum=1, sesudah=1

Tiga worker memperebutkan sepuluh job dalam urutan yang berbeda setiap eksekusi, tetapi cetakan selalu sama: hasil dikumpulkan dulu, diurutkan dengan slices.SortFunc berdasarkan JobID, baru dicetak. Baris terakhir menghitung goroutine sebelum pool menyala dan sesudahnya mati — angka yang sama berarti tidak ada pekerja yang lupa pulang. Penutupan channel hasil itu sendiri masih berupa kotak hitam pool.Shutdown(); Bagian 3 membongkarnya.

Rujukan Materi #

Goroutine — cara Go menjadwalkan fungsi sebagai goroutine.
Channel — semantik kirim-terima dan channel tanpa buffer.
Sync WaitGroup — pola Add sebelum go dan Wait untuk menunggu.
Dokumentasi sync — referensi resmi WaitGroup.

Worker hidup dan job mengalir. Satu pertanyaan masih terbuka: siapa yang menutup channel hasil, dan bagaimana pool berhenti dengan bersih? Bagian 3 menjawabnya dengan pengawas, Shutdown, dan context.

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 2: Implementasi Goroutine

Goroutine Worker #

Menyalakan Pool #

Mengirim Job dengan Backpressure #

Demo Distribusi Sepuluh Job #

Rujukan Materi #

Berkas Sumber

main.go

pool_test.go

pool.go

worker_test.go

worker.go