Concurrency patterns: pipeline & cancellation

Rangkai tahap goroutine dan batalkan lewat context.

Pahami konsepnya

Pipeline adalah goroutine-goroutine yang terhubung channel: tahap pertama MEMBUAT data (generator), tahap-tahap berikut MENGOLAH dan meneruskan hasilnya lewat channel keluaran. Tiap tahap berjalan bersamaan — sambil satu tahap menghitung, yang lain sudah menyiapkan berikutnya. Kunci polanya: setiap tahap menutup channel keluarannya setelah selesai, dan for range di hilir otomatis berhenti.

Fan-in/fan-out memperbanyak jalur: beberapa tahap pekerja yang SAMA membaca dari satu channel input (fan-out — memparalelkan kerja), hasilnya digabung ke satu channel (fan-in). Dengan begitu pekerjaan lambat dibagi ke semua core tanpa mengubah bentuk kodenya.

Pembatalan elegan lewat context.Context: context.WithCancel membuat channel Done yang tertutup saat cancel dipanggil. Setiap tahap memilih antara channel kerja dan ctx.Done — begitu sinyal pembatalan tiba, seluruh pipeline angkat kaki dengan rapi. Varian WithTimeout memasang pembatalan otomatis berbasis waktu; pola ini wajib di API modern Go.

Coba contoh kode

package main

import "fmt"

func generator(angka ...int) <-chan int {
    keluar := make(chan int)
    go func() {
        defer close(keluar)
        for _, n := range angka {
            keluar <- n
        }
    }()
    return keluar
}

func kuadrat(masuk <-chan int) <-chan int {
    keluar := make(chan int)
    go func() {
        defer close(keluar)
        for n := range masuk {
            keluar <- n * n
        }
    }()
    return keluar
}

func main() {
    for hasil := range kuadrat(generator(1, 2, 3, 4)) {
        fmt.Println(hasil) // 1 4 9 16
    }
}

Dua tahap terangkai: generator mengirim 1-4 lalu menutup; kuadrat menerima lewat range, mengalikan, meneruskan, lalu ikut menutup. main hanya membaca hasil akhir — penutupan berantai membuat seluruh pipeline mati dengan rapi setelah data habis. Tanda panah pada tipe (<-chan) membatasi channel jadi hanya-keluar: konsumen tak bisa mengirim balik.

Pembatalan dengan context

func kerja(ctx context.Context) <-chan int {
    keluar := make(chan int)
    go func() {
        defer close(keluar)
        for i := 1; ; i++ {
            select {
            case <-ctx.Done():
                return // sinyal batal: keluar rapi
            case keluar <- i:
            }
        }
    }()
    return keluar
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    hasil := kerja(ctx)

    fmt.Println(<-hasil) // 1
    fmt.Println(<-hasil) // 2
    cancel()             // cukup: hentikan generator
}

select menjunjung dua kemungkinan: kirim nilai berikutnya, atau berhenti saat ctx.Done tertutup. cancel() menutup Done seketika dan generator pulang — goroutine tidak menggantung menunggu konsumen yang sudah pergi. context.WithTimeout(ctx, 2*time.Second) memakai bentuk yang sama dengan alarm otomatis.

Latihan singkat

Uji: sambung pipeline filter-genap → kuadrat, terima tiga hasil lalu batalkan — harus `4`, `16`, `36`, `selesai` (empat baris).