Golang Concurrency #8

Berikut adalah contoh implementasi worker pool menggunakan concurrency di Golang yang efisien. Contoh ini menggunakan beberapa worker untuk memproses serangkaian pekerjaan (jobs) secara paralel.

---

Kode Worker Pool

package main

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

// Struktur Job, mewakili sebuah pekerjaan dengan ID dan nilai.
type Job struct {
	id    int
	value int
}

// Struktur Result, menyimpan hasil pemrosesan job.
type Result struct {
	job    Job
	output int
}

// Fungsi worker yang memproses job dari channel jobs dan mengirim hasilnya ke channel results.
func worker(workerID int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for job := range jobs {
		fmt.Printf("Worker %d mulai memproses job %d\n", workerID, job.id)
		// Simulasi proses dengan delay
		time.Sleep(500 * time.Millisecond)
		// Misal proses: menghitung kuadrat dari nilai job.
		output := job.value * job.value
		results <- Result{job: job, output: output}
		fmt.Printf("Worker %d selesai job %d\n", workerID, job.id)
	}
}

func main() {
	numJobs := 10    // Jumlah total job yang akan diproses
	numWorkers := 3  // Jumlah worker yang berjalan secara konkuren

	// Membuat channel untuk job dan results. Menggunakan buffer sesuai jumlah job.
	jobs := make(chan Job, numJobs)
	results := make(chan Result, numJobs)

	var wg sync.WaitGroup

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

	// Mengirimkan job ke channel jobs.
	for j := 1; j <= numJobs; j++ {
		jobs <- Job{id: j, value: j}
	}
	close(jobs) // Menandakan tidak ada lagi job yang dikirimkan

	// Menunggu semua worker selesai memproses.
	wg.Wait()
	close(results) // Setelah semua worker selesai, tutup channel results

	// Membaca dan menampilkan semua hasil.
	for result := range results {
		fmt.Printf("Job %d dengan nilai %d menghasilkan output %d\n", result.job.id, result.job.value, result.output)
	}
}

---

Penjelasan Kode

1. Definisi Struktur Data:

   • Job: Mewakili sebuah pekerjaan dengan properti id dan value.
   • Result: Menyimpan hasil dari pemrosesan sebuah job, termasuk job itu sendiri dan hasil perhitungannya.

2. Fungsi Worker:

   • Setiap worker berjalan di goroutine terpisah.
   • Worker mengambil job dari channel jobs dengan menggunakan loop for range. Karena jobs adalah channel, loop ini akan berjalan sampai channel ditutup.
   • Setelah menyelesaikan sebuah job, worker mengirimkan hasilnya ke channel results.
   • defer wg.Done() memastikan bahwa setiap worker mengurangi counter WaitGroup setelah selesai.

3. Fungsi main:

   • Mempersiapkan channel jobs dan results dengan buffer yang sesuai untuk menghindari deadlock.
   • Meluncurkan sejumlah worker (dalam contoh ini 3) yang akan mendengarkan channel jobs.
   • Mengirimkan sejumlah job ke channel jobs dan kemudian menutup channel tersebut agar worker tahu bahwa tidak ada job baru.
   • Menunggu semua worker selesai dengan wg.Wait(), lalu menutup channel results agar loop pembacaan di bagian akhir bisa berhenti.
   • Membaca dan menampilkan hasil dari channel results dengan loop for range.

---

Mengapa Pendekatan Ini Efisien?

• Concurrency:
  Worker bekerja secara paralel, sehingga pekerjaan dapat diproses bersamaan dan mengurangi total waktu pemrosesan.

• Penggunaan Channel:
  Channel menjamin komunikasi yang aman antar goroutine tanpa perlu sinkronisasi manual (misalnya menggunakan mutex).

• Worker Pool:
  Worker pool membantu membatasi jumlah goroutine yang berjalan secara bersamaan, menghindari overhead pembuatan goroutine yang berlebihan.

• Sinkronisasi dengan WaitGroup:
  Menggunakan sync.WaitGroup memastikan bahwa program menunggu sampai semua worker selesai sebelum menutup channel dan melanjutkan pemrosesan hasil.

Contoh ini merupakan salah satu pola concurrency yang umum dan efisien di Golang. Jika ada pertanyaan lebih lanjut atau ingin eksplorasi contoh lain, silakan tanyakan!