Golang Concurrency #9

Februari 21, 2025

Mari kita bahas mekanismenya langkah demi langkah.

1. Pembuatan Worker dan Pengiriman Job

Pembuatan Worker:
Pada kode, kita membuat 3 worker (masing-masing berjalan di goroutine terpisah) dengan melakukan:
```
for w := 1; w <= numWorkers; w++ {
    wg.Add(1)
    go worker(w, jobs, results, &wg)
}
```
Masing-masing worker menjalankan loop:
```
for job := range jobs { ... }
```
Worker akan menunggu (melakukan blocking read) dari channel jobs sampai ada data.
Pengiriman Job:
Kemudian, di goroutine utama, kita mengirimkan 10 job ke channel jobs:
```
for j := 1; j <= numJobs; j++ {
    jobs <- Job{id: j, value: j}
}
close(jobs)
```
Karena channel jobs dibuat dengan buffer sebesar numJobs (misal make(chan Job, numJobs)), semua job dapat dimasukkan tanpa harus menunggu worker membaca. Jadi, pengiriman job tidak akan blocking selama buffer belum penuh. Bahkan jika buffer sudah penuh, worker yang aktif membaca dari channel akan membuat ruang sehingga pengiriman berikutnya bisa diterima.

2. Bagaimana Job Didistribusikan ke 3 Worker?

Channel jobs berperan sebagai antrian bersama. Ketika sebuah worker melakukan for job := range jobs:

Blocking Read:
Jika channel kosong, worker akan menunggu sampai ada job yang dikirim.
Distribusi Job:
Begitu ada job, salah satu worker yang sedang menunggu akan membaca job tersebut. Jika beberapa worker menunggu, job pertama akan diambil oleh salah satu worker secara tidak terduga (bersifat non-deterministik), dan begitu seterusnya. Dengan begitu, job-job didistribusikan secara concurrent ke 3 worker.

3. Mekanisme `wg.Wait()` dan `close(results)`

Dalam contoh ini, kode untuk wg.Wait() dan close(results) diletakkan di goroutine utama, tepat setelah pengiriman job:

// Di goroutine utama setelah mengirim job:
wg.Wait()      // Menunggu semua worker selesai
close(results) // Setelah worker selesai, channel results ditutup

Mengapa tidak ditempatkan di goroutine terpisah?

Desain Sederhana:
Karena kita ingin memastikan bahwa semua worker sudah selesai memproses job sebelum kita mulai membaca hasil. Jadi, kita ingin:
1. Mengirim job.
2. Menunggu worker selesai (wg.Wait()).
3. Menutup channel hasil.
4. Membaca semua hasil.
Dengan cara ini, urutan eksekusi terjamin.
Tidak Mengganggu Proses Worker:
wg.Wait() hanya memblokir goroutine utama. Worker-worker yang berjalan secara concurrent tidak terpengaruh karena mereka berada di goroutine masing-masing. Jadi, walaupun goroutine utama sedang menunggu, worker tetap membaca dan memproses job dari channel jobs.
Channel results:
Hasil-hasil yang dikirim ke channel results oleh worker disimpan dalam buffer (karena kita juga membuat channel results dengan kapasitas cukup). Walaupun goroutine utama belum segera membaca, data tetap tersimpan dalam buffer channel.
Setelah wg.Wait() selesai, kita memanggil close(results). Hal ini penting agar loop:
```
for result := range results { ... }
```
tahu kapan harus berhenti membaca (yaitu, ketika channel sudah kosong dan tertutup).

4. Skema Kerja Keseluruhan

Worker Dimulai:
Tiga worker siap menerima job dari channel jobs.
Job Dikirim:
Goroutine utama mengirim 10 job ke channel jobs. Karena channel dibuffer, pengiriman tidak akan blocking.
Distribusi & Pemrosesan Job:
Worker secara concurrent membaca dari channel jobs. Masing-masing worker memproses job yang didapat, lalu mengirimkan hasilnya ke channel results.
Menunggu Selesai:
Goroutine utama memanggil wg.Wait() dan menunggu sampai semua worker selesai (yakni, semua worker telah memanggil wg.Done()).
Menutup Channel Hasil:
Setelah semua worker selesai, goroutine utama menutup channel results sehingga loop pembacaan di akhir tahu bahwa tidak akan ada data baru.
Membaca Hasil:
Goroutine utama membaca semua hasil dari channel results dan memproses atau menampilkannya.

5. Jawaban Langsung ke Pertanyaan

Apakah Pengiriman Job Menjadi Blocking?
Karena channel jobs dibuffer dengan ukuran yang cukup, pengiriman job tidak akan blocking selama buffer tidak penuh. Selain itu, worker yang sudah berjalan secara concurrent juga membaca job sehingga buffer tidak cepat penuh.
Kemana Job Dikirim?
Job yang dikirim disimpan dalam buffer channel jobs dan segera akan diambil oleh salah satu dari 3 worker secara concurrent melalui loop for job := range jobs.
Kenapa wg.Wait() dan close(results) Diletakkan di Goroutine Utama?
Karena kita ingin alur eksekusi yang terstruktur:
- Pertama, mengirim semua job.
- Kedua, menunggu sampai semua worker selesai.
- Ketiga, menutup channel hasil dan kemudian membaca hasil.
Menempatkannya di goroutine utama memastikan bahwa kita tidak mulai membaca hasil sampai semua pekerjaan selesai. Jika dipindahkan ke goroutine terpisah, kita bisa saja memproses hasil secara real-time sambil menunggu worker selesai, tapi itu adalah desain yang berbeda.
Dalam desain ini, kita memilih kesederhanaan: tunggu semuanya selesai, baru baca hasil.

Semoga penjelasan ini membantu memperjelas mekanisme distribusi job, blocking pada channel, serta alasan penempatan wg.Wait() dan close(results) di goroutine utama. Jika masih ada pertanyaan atau butuh penjelasan tambahan, silakan ditanyakan!