Pemrosesan data sering berbentuk rangkaian: data mentah dibaca, disaring, diubah, lalu dicetak. Bila setiap tahap berjalan satu per satu, tahap pembaca menganggur saat tahap pengubahan bekerja, dan sebaliknya. Pola pipeline menyusun setiap tahap sebagai goroutine yang dihubungkan channel, sehingga semua tahap bekerja bersamaan seperti sabuk produksi: generator mengirim angka pertama sambil tahap berikutnya menunggu angka kedua. Chapter ini membangun pipeline sederhana dari pola blog resmi Go — generate, transform, sink — beserta aturan penutupan channel yang membuat seluruh rangkaian berhenti dengan rapi.

Tahapan yang Terhubung Channel #

Pipeline di Go tersusun dari tahapan-tahapan: setiap tahap adalah sekelompok goroutine yang menjalankan fungsi yang sama, menerima nilai dari channel hulu lewat pernyataan range, lalu mengirim nilai hasil ke channel hilir. Bila satu tahap diisi satu goroutine, urutan nilai terjaga dari hulu ke hilir karena satu channel adalah antrean yang patuh urutan.

Implementasi Dua Tahap #

Berkas pipeline-demo.go membangun tahap generator yang mengubah kumpulan angka menjadi aliran nilai, dan tahap kuadrat yang mengalikan setiap angka dengan dirinya sendiri:

go 
package main

import (
	"fmt"
)

// generator mengubah kumpulan angka menjadi stream di sebuah channel.
func generator(angka ...int) <-chan int {
	keluar := make(chan int)
	go func() {
		defer close(keluar)
		for _, n := range angka {
			keluar <- n
		}
	}()
	return keluar
}

// kuadrat membaca setiap angka dari channel masukan, menghitung
// kuadratnya, lalu mengirim hasilnya ke channel keluaran.
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() {
	sumber := generator(1, 2, 3, 4, 5)
	hasil := kuadrat(sumber)

	// Tahap akhir: kumpulkan hasil sampai channel ditutup.
	terkumpul := []int{}
	for v := range hasil {
		terkumpul = append(terkumpul, v)
	}
	fmt.Println("hasil pipeline:", terkumpul)
}

Jalankan program tersebut dari direktori chapter:

bash 
go run pipeline-demo.go

Output yang Dihasilkan:

text 
hasil pipeline: [1 4 9 16 25]

Mari kita telaah baris demi baris program di atas:

  1. Baris 8–17 (generator): menerima sejumlah angka (angka ...int), membuat channel keluar, lalu menjalankan goroutine yang mengirim angka satu per satu; nilai baliknya bertipe <-chan int — channel khusus menerima — agar tahap lain tidak bisa mengirim ke hulu.
  2. Baris 11 (defer close(keluar)): channel ditutup begitu goroutine selesai mengirim; penutupan ini menjadi sinyal “aliran selesai” bagi tahap hilir.
  3. Baris 21–30 (kuadrat): tahap kedua menjalankan for n := range masuk; perulangan menerima nilai selama channel terbuka dan berakhir begitu channel tertutup, lalu goroutine ini menutup channel keluarannya sendiri.
  4. Baris 33–34 (sumber dan hasil): kedua tahap dirangkai — generator mengalir ke kuadrat; mulai titik ini kedua goroutine bekerja bersamaan.
  5. Baris 37–41 (tahap akhir): main menjadi tahap akhir yang mengumpulkan nilai sampai channel hasil ditutup; perulangan for v := range hasil itulah sinkronisasi akhir pipeline — program selesai hanya setelah seluruh aliran habis.

Hasil [1 4 9 16 25] selalu terurut karena setiap tahap hanya berisi satu goroutine: generator mengirim 1 sebelum 2, kuadrat memproses 1 sebelum 2, dan antrean channel menjaga urutan itu di setiap sambungan.

Penutupan Berjenjang #

Perhatikan bahwa tidak ada satu pun tahap yang diberi tahu “berhentilah” secara langsung. Penutupan justru berjalan sendiri dari hulu ke hilir secara berjenjang: generator selesai mengirim dan menutup channelnya; kuadrat yang sedang menjalankan range masuk melihat penutupan itu, keluar dari perulangan, lalu menutup channel keluarannya; tahap akhir melihat penutupan kedua dan keluar dari pengumpulan. Aturan praktisnya: pengirim selalu bertanggung jawab menutup channel, dan penutupan menyebar mengikuti arah aliran data.

Skala Lebih Besar: Fan-out dan Fan-in #

Pipeline satu-goroutine-per-tahap menjaga urutan tetapi membatasi lewatnya satu nilai pada satu waktu per tahap. Blog resmi Go memperluas pola ini dengan dua teknik: fan-out — beberapa goroutine menjalankan tahap yang sama membaca satu channel hulu — dan fan-in — satu tahap penggabung membaca banyak channel dan menyatukan kembali alirannya:

go 
// Fan-out: dua tahap kuadrat membaca channel yang sama.
k1 := kuadrat(sumber)
k2 := kuadrat(sumber)
// Fan-in: fungsi merge menggabungkan k1 dan k2 ke satu channel keluaran.
hasil := merge(k1, k2)

Dengan fan-out, pekerjaan tahap yang mahal dibagi ke banyak prosesor; urutan nilai, tentu saja, tidak lagi terjamin dan pengurutan kembali menjadi tanggung jawaba tahap akhir — sama seperti pada worker pool.

Dari sini fondasi concurrency kita sudah lengkap: goroutine, channel, select, sync, context, dan dua pola penyusunan kerja. Chapter berikutnya menghadirkan sisi yang kelam dari kebebasan ini — data race — beserta alat resminya untuk mendeteksinya.

Referensi #

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

pipeline-demo.go

Unduh berkas
go 
package main

import (
	"fmt"
)

// generator mengubah kumpulan angka menjadi stream di sebuah channel.
func generator(angka ...int) <-chan int {
	keluar := make(chan int)
	go func() {
		defer close(keluar)
		for _, n := range angka {
			keluar <- n
		}
	}()
	return keluar
}

// kuadrat membaca setiap angka dari channel masukan, menghitung
// kuadratnya, lalu mengirim hasilnya ke channel keluaran.
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() {
	sumber := generator(1, 2, 3, 4, 5)
	hasil := kuadrat(sumber)

	// Tahap akhir: kumpulkan hasil sampai channel ditutup.
	terkumpul := []int{}
	for v := range hasil {
		terkumpul = append(terkumpul, v)
	}
	fmt.Println("hasil pipeline:", terkumpul)
}