Producer-Consumer Pattern #

Ketika sebuah sistem perlu memproses data secara terus-menerus — entah itu antrian email, stream event dari sensor IoT, atau batch job nightly — hampir selalu ada dua tanggung jawab yang terlibat: menghasilkan data dan memproses data. Masalah muncul ketika kedua tanggung jawab ini dikerjakan dalam satu alur yang tightly coupled: producer harus menunggu consumer selesai, atau consumer terus di-polling padahal belum ada data baru. Akibatnya throughput rendah, latensi tinggi, dan kode sulit diuji. Producer-Consumer Pattern hadir untuk memecah dua tanggung jawab itu menjadi unit yang berdiri sendiri dan berkomunikasi melalui buffer bersama, sehingga keduanya bisa berjalan concurrently tanpa saling blocking secara langsung.

Apa itu Producer-Consumer Pattern? #

Producer-Consumer Pattern adalah concurrency pattern yang memisahkan proses pembuatan data dari proses konsumsinya melalui sebuah antrian atau buffer bersama.

Tiga komponen inti dalam pola ini adalah:

Alur komunikasinya selalu satu arah:

flowchart LR
    P1([Producer 1]) --> B[(Buffer / Channel)]
    P2([Producer 2]) --> B
    B --> C1([Consumer 1])
    B --> C2([Consumer 2])
    B --> C3([Consumer 3])

Producer dan consumer berjalan secara independen. Producer tidak tahu siapa yang akan mengonsumsi datanya, dan consumer tidak tahu bagaimana data itu dihasilkan. Yang mereka tahu hanya satu hal: ada buffer di antara mereka.

Di Golang, buffer ini direpresentasikan secara native menggunakan channel — sebuah primitif concurrency yang thread-safe, sudah ada di runtime, dan idiomatik untuk komunikasi antar goroutine.

Mengapa Channel, Bukan Mutex? #

Sebelum masuk ke implementasi, penting untuk memahami mengapa Golang mendorong channel sebagai mekanisme komunikasi, bukan mutex atau shared memory biasa.

Prinsip Golang yang terkenal berbunyi: “Do not communicate by sharing memory; instead, share memory by communicating.” Artinya, daripada beberapa goroutine mengakses satu variabel bersama dan kamu harus pasang lock di sana-sini, lebih baik goroutine itu saling kirim dan terima data melalui channel.

Pertimbangkan perbedaan pendekatannya:

// ANTI-PATTERN: shared memory dengan mutex — rawan data race jika lupa lock
var queue []int
var mu sync.Mutex

func produce(val int) {
    mu.Lock()
    queue = append(queue, val)
    mu.Unlock()
}

func consume() int {
    mu.Lock()
    defer mu.Unlock()
    if len(queue) == 0 {
        return -1 // tidak ada cara idiomatik untuk blocking di sini
    }
    val := queue[0]
    queue = queue[1:]
    return val
}

// BENAR: channel sebagai buffer — blocking secara otomatis, tidak perlu mutex manual
ch := make(chan int, 10)

go func() { ch <- produceValue() }()  // producer
go func() { process(<-ch) }()         // consumer

Channel memberikan blocking semantics secara gratis: producer akan otomatis menunggu jika buffer penuh, dan consumer akan otomatis menunggu jika buffer kosong. Tidak perlu sync.Cond, tidak perlu polling loop.

Single Producer, Single Consumer #

Implementasi paling sederhana adalah satu goroutine producer dan satu goroutine consumer. Ini titik awal yang baik untuk memahami mekanisme dasarnya.

package main

import (
	"fmt"
	"time"
)

// producer mengirim data ke channel dan menutupnya saat selesai
func producer(ch chan<- int) {
	for i := 1; i <= 5; i++ {
		fmt.Printf("[Producer] Menghasilkan item: %d\n", i)
		ch <- i
		time.Sleep(300 * time.Millisecond)
	}
	close(ch) // BENAR: channel ditutup dari sisi producer
}

// consumer membaca dari channel sampai ditutup
func consumer(ch <-chan int) {
	for item := range ch { // range otomatis berhenti saat channel ditutup
		fmt.Printf("[Consumer] Memproses item: %d\n", item)
		time.Sleep(700 * time.Millisecond)
	}
	fmt.Println("[Consumer] Selesai")
}

func main() {
	ch := make(chan int, 3) // buffered channel, kapasitas 3

	go producer(ch)
	consumer(ch) // consumer berjalan di goroutine utama
}

Ada beberapa detail penting di sini:

Pertama, channel dideklarasikan dengan kapasitas 3. Ini artinya producer bisa memasukkan sampai 3 item sebelum blocking, memberi producer “ruang bernapas” ketika consumer sedang sibuk memproses item sebelumnya.

Kedua, close(ch) dipanggil dari producer, bukan consumer. Ini aturan baku di Golang: hanya pengirim yang boleh menutup channel.

Ketiga, for item := range ch di consumer secara otomatis berhenti saat channel ditutup dan semua item sudah terbaca — tidak perlu kondisi terminasi manual.

Alur eksekusinya bisa divisualisasikan seperti ini:

sequenceDiagram
    participant P as Producer
    participant CH as Channel (buf=3)
    participant C as Consumer

    P->>CH: send(1)
    P->>CH: send(2)
    P->>CH: send(3)
    Note over P,CH: Buffer penuh, producer blocking
    C->>CH: receive() → 1
    Note over CH,C: Consumer mulai proses item 1
    P->>CH: send(4)
    C->>CH: receive() → 2
    P->>CH: send(5)
    P->>CH: close()
    C->>CH: receive() → 3
    C->>CH: receive() → 4
    C->>CH: receive() → 5
    Note over C: Channel ditutup, range selesai

Multiple Producer & Multiple Consumer #

Kekuatan sesungguhnya dari Producer-Consumer Pattern baru terlihat ketika kamu menskalakan ke banyak producer dan banyak consumer. Golang membuat ini sangat mudah karena channel sudah thread-safe — banyak goroutine bisa menulis dan membaca dari channel yang sama tanpa race condition.

package main

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

func producer(id int, ch chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 1; i <= 4; i++ {
		item := fmt.Sprintf("P%d-item%d", id, i)
		fmt.Printf("[Producer %d] Menghasilkan: %s\n", id, item)
		ch <- item
		time.Sleep(200 * time.Millisecond)
	}
}

func consumer(id int, ch <-chan string, wg *sync.WaitGroup) {
	defer wg.Done()
	for item := range ch {
		fmt.Printf("[Consumer %d] Memproses: %s\n", id, item)
		time.Sleep(500 * time.Millisecond)
	}
}

func main() {
	const numProducers = 3
	const numConsumers = 2
	const bufferSize = 10

	ch := make(chan string, bufferSize)

	var prodWG sync.WaitGroup
	var consWG sync.WaitGroup

	// Jalankan consumers lebih dulu agar siap menerima data
	for i := 1; i <= numConsumers; i++ {
		consWG.Add(1)
		go consumer(i, ch, &consWG)
	}

	// Jalankan producers
	for i := 1; i <= numProducers; i++ {
		prodWG.Add(1)
		go producer(i, ch, &prodWG)
	}

	// Tunggu semua producer selesai, lalu tutup channel
	prodWG.Wait()
	close(ch)

	// Tunggu semua consumer selesai memproses sisa data
	consWG.Wait()

	fmt.Println("Semua item berhasil diproses")
}

Urutan operasi di sini penting dan sering menjadi sumber bug jika salah:

flowchart TD
    A[Start consumers] --> B[Start producers]
    B --> C[prodWG.Wait\nmenunggu semua producer selesai]
    C --> D[close ch\nmenutup channel]
    D --> E[consWG.Wait\nmenunggu consumer habiskan sisa item]
    E --> F[Program selesai]

Jangan tutup channel sebelum semua producer selesai. Jika ada producer yang masih berjalan saat close(ch) dipanggil, program akan panic dengan pesan send on closed channel. Selalu gunakan sync.WaitGroup untuk memastikan semua producer sudah selesai sebelum channel ditutup.

Memahami Backpressure #

Backpressure adalah mekanisme di mana consumer yang lambat secara otomatis “menekan balik” producer agar tidak menghasilkan data lebih cepat dari yang bisa diproses. Ini adalah salah satu keunggulan terbesar dari Producer-Consumer Pattern berbasis channel.

Ketika buffer penuh, operasi ch <- data di producer akan blocking sampai ada consumer yang mengambil satu item. Ini bukan bug — ini adalah desain yang tepat untuk menjaga sistem tetap stabil di bawah beban tinggi.

// ANTI-PATTERN: unbuffered channel dengan goroutine tanpa batas
// setiap item langsung dapat goroutine baru — bisa exhausted memory
func processAll(items []int) {
	ch := make(chan int) // unbuffered
	for _, item := range items {
		go func(v int) { ch <- v }() // ✗ goroutine leak jika tidak ada consumer
	}
}

// BENAR: bounded buffer dengan jumlah consumer tetap
func processAll(items []int) {
	ch := make(chan int, 50) // buffer terbatas
	var wg sync.WaitGroup

	// Consumer pool tetap
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for item := range ch {
				process(item) // ✓ jumlah consumer tetap, backpressure alami
			}
		}()
	}

	for _, item := range items {
		ch <- item // blocking saat buffer penuh — backpressure bekerja
	}

	close(ch)
	wg.Wait()
}

Panduan memilih ukuran buffer:

Cancellation dengan Context #

Sistem yang berjalan lama — seperti background job processor atau daemon — perlu bisa dihentikan dengan bersih. Pola idiomatik di Golang adalah menggunakan context.Context sebagai sinyal pembatalan.

package main

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

func producer(ctx context.Context, ch chan<- int) {
	defer close(ch)
	i := 0
	for {
		select {
		case <-ctx.Done():
			fmt.Println("[Producer] Context dibatalkan, berhenti produksi")
			return
		case ch <- i:
			fmt.Printf("[Producer] Menghasilkan: %d\n", i)
			i++
			time.Sleep(200 * time.Millisecond)
		}
	}
}

func consumer(ctx context.Context, id int, ch <-chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			fmt.Printf("[Consumer %d] Context dibatalkan, berhenti konsumsi\n", id)
			return
		case item, ok := <-ch:
			if !ok {
				fmt.Printf("[Consumer %d] Channel ditutup, selesai\n", id)
				return
			}
			fmt.Printf("[Consumer %d] Memproses: %d\n", id, item)
			time.Sleep(500 * time.Millisecond)
		}
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	ch := make(chan int, 5)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go consumer(ctx, i, ch, &wg)
	}

	go producer(ctx, ch)

	wg.Wait()
	fmt.Println("Semua consumer selesai")
}

Penggunaan select dengan dua case — ctx.Done() dan operasi channel — memungkinkan producer dan consumer responsif terhadap sinyal pembatalan tanpa perlu polling atau variabel boolean yang di-share.

stateDiagram-v2
    [*] --> Running : Start goroutine
    Running --> Processing : Terima item dari channel
    Processing --> Running : Item selesai diproses
    Running --> Cancelled : ctx.Done() menerima sinyal
    Running --> Done : Channel ditutup, ok == false
    Cancelled --> [*]
    Done --> [*]

Pipeline: Menyambungkan Producer-Consumer #

Pola Producer-Consumer bisa dikembangkan menjadi pipeline — serangkaian stage di mana output satu stage menjadi input stage berikutnya. Ini sangat berguna untuk pemrosesan data yang punya beberapa tahap transformasi.

package main

import (
	"fmt"
	"strings"
)

// Stage 1: hasilkan kata-kata
func generate(words ...string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for _, w := range words {
			ch <- w
		}
	}()
	return ch
}

// Stage 2: ubah ke huruf kapital
func toUpper(in <-chan string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for w := range in {
			ch <- strings.ToUpper(w)
		}
	}()
	return ch
}

// Stage 3: tambahkan prefix
func addPrefix(prefix string, in <-chan string) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for w := range in {
			ch <- fmt.Sprintf("[%s] %s", prefix, w)
		}
	}()
	return ch
}

func main() {
	// Sambungkan pipeline: generate → toUpper → addPrefix
	words := generate("golang", "concurrency", "channel", "pipeline")
	upper := toUpper(words)
	result := addPrefix("OUTPUT", upper)

	for item := range result {
		fmt.Println(item)
	}
}

Hasilnya:

[OUTPUT] GOLANG
[OUTPUT] CONCURRENCY
[OUTPUT] CHANNEL
[OUTPUT] PIPELINE

Setiap stage berjalan sebagai goroutine terpisah dan berkomunikasi melalui channel. Ini adalah contoh nyata bagaimana Producer-Consumer menjadi fondasi dari pola yang lebih kompleks.

flowchart LR
    A[generate\nwords] -->|chan string| B[toUpper]
    B -->|chan string| C[addPrefix]
    C -->|chan string| D[main\nconsumer]

Producer-Consumer vs Worker Pool #

Kedua pola ini sering disebut bersama dan memang berkaitan erat, tapi punya fokus yang berbeda.

Dalam praktiknya, Worker Pool adalah spesialisasi dari Producer-Consumer: kamu punya producer yang mengisi antrian, dan pool worker (consumer) dengan jumlah tetap yang mengambil dari antrian itu. Memahami Producer-Consumer dengan baik adalah prasyarat untuk memahami Worker Pool.

Anti-Pattern yang Harus Dihindari #

// ✗ Menutup channel dari consumer — akan panic jika producer masih send
func badConsumer(ch chan int) {
    val := <-ch
    close(ch) // JANGAN: hanya producer yang boleh menutup channel
    process(val)
}

// ✓ Tutup channel hanya dari producer, consumer hanya membaca
func goodConsumer(ch <-chan int) { // <-chan: read-only, tidak bisa close
    for val := range ch {
        process(val)
    }
}

// ✗ Goroutine leak: consumer tidak punya jalur exit
func leakySetup() {
    ch := make(chan int)
    go func() {
        for val := range ch { // goroutine ini tidak pernah selesai
            process(val)
        }
    }()
    // ch tidak pernah ditutup → goroutine leak selamanya
}

// ✓ Selalu pastikan channel akan ditutup
func safeSetup() chan int {
    ch := make(chan int, 10)
    go func() {
        defer close(ch) // ✓ deferred close memastikan channel selalu ditutup
        for _, item := range fetchItems() {
            ch <- item
        }
    }()
    return ch
}

// ✗ Buffer tidak terbatas via goroutine baru per item
func unboundedProducer(items []int) {
    ch := make(chan int)
    for _, item := range items { // 1 juta item = 1 juta goroutine
        go func(v int) { ch <- v }() // ✗ bisa exhausted memory
    }
}

// ✓ Gunakan buffered channel dengan ukuran yang wajar
func boundedProducer(items []int) <-chan int {
    ch := make(chan int, 100) // ✓ buffer terbatas, backpressure alami
    go func() {
        defer close(ch)
        for _, item := range items {
            ch <- item
        }
    }()
    return ch
}

// ✗ Tidak menangani panic di consumer — bisa kill seluruh program
func unsafeConsumer(ch <-chan int) {
    for val := range ch {
        riskyOperation(val) // jika panic, program crash
    }
}

// ✓ Recover panic di consumer agar goroutine pool tetap hidup
func safeConsumerWithRecover(ch <-chan int) {
    for val := range ch {
        func() {
            defer func() {
                if r := recover(); r != nil {
                    log.Printf("Recovered panic: %v", r) // ✓ log dan lanjutkan
                }
            }()
            riskyOperation(val)
        }()
    }
}

Checklist Review Producer-Consumer #

DESAIN CHANNEL:
  □ Channel bertipe directional (chan<- untuk producer, <-chan untuk consumer)
  □ Ukuran buffer dipilih berdasarkan rasio kecepatan producer vs consumer
  □ Channel hanya ditutup dari sisi producer

LIFECYCLE GOROUTINE:
  □ Setiap goroutine punya jalur exit yang jelas (context, channel close, atau done signal)
  □ Tidak ada goroutine yang bisa leak (channel yang tidak pernah ditutup)
  □ WaitGroup digunakan untuk sinkronisasi selesainya goroutine

CANCELLATION:
  □ Context digunakan untuk sinyal pembatalan pada sistem jangka panjang
  □ select digunakan untuk responsif terhadap ctx.Done() dan channel secara bersamaan
  □ Timeout dipertimbangkan untuk operasi yang bisa macet

ERROR HANDLING:
  □ Panic di consumer di-recover agar tidak membunuh seluruh program
  □ Error dari consumer dikomunikasikan kembali (misal via error channel terpisah)
  □ Partial failure di multi-producer/consumer ditangani dengan benar

OBSERVABILITY:
  □ Ada logging di setiap stage untuk debugging
  □ Metrik panjang buffer dan throughput dimonitor
  □ Goroutine count dipantau untuk deteksi leak

Ringkasan #

• Producer-Consumer memisahkan dua tanggung jawab — menghasilkan data dan memproses data — sehingga keduanya bisa berjalan concurrent tanpa coupling langsung.
• Channel adalah buffer idiomatik di Golang — thread-safe, mendukung blocking semantics, dan mengeliminasi kebutuhan mutex manual untuk komunikasi antar goroutine.
• Hanya producer yang boleh menutup channel — menutup channel dari consumer akan menyebabkan panic jika producer masih send.
• for item := range ch adalah cara idiomatik membaca channel — otomatis berhenti saat channel ditutup dan semua item habis terbaca.
• sync.WaitGroup wajib digunakan untuk memastikan semua producer selesai sebelum channel ditutup, dan semua consumer selesai sebelum program exit.
• Backpressure bekerja secara otomatis — channel yang penuh akan memblocking producer, menjaga sistem tetap stabil tanpa logika throttling manual.
• Gunakan context.Context untuk cancellation pada sistem jangka panjang — select dengan ctx.Done() memungkinkan goroutine responsif terhadap sinyal berhenti.
• Pipeline adalah evolusi alami dari Producer-Consumer — output satu stage menjadi input stage berikutnya, masing-masing berjalan sebagai goroutine terpisah.
• Worker Pool adalah spesialisasi Producer-Consumer dengan jumlah consumer tetap (bounded) — memahami Producer-Consumer adalah prasyarat memahami Worker Pool.
• Recover panic di consumer agar satu item yang gagal tidak membunuh seluruh goroutine pool.

← Sebelumnya: Worker Pool   Berikutnya: Future-Promise →