Mengontrol Aliran Data dengan Stream di Golang

Dalam era sistem yang serba real-time dan berskala besar, mengelola aliran data menjadi kebutuhan penting di hampir setiap aplikasi—mulai dari backend yang menangani request ribuan pengguna, proses ETL data besar, sampai aplikasi IoT yang datanya terus mengalir. Salah satu konsep fundamental yang membantu developer dalam mengatasi masalah ini adalah stream.

Artikel ini akan membahas bagaimana mengontrol aliran data menggunakan stream di Go: mulai dari mengapa stream penting, bagaimana prinsip kerjanya, hingga implementasi teknik seperti buffering, throttling, dan backpressure.

Mengapa Harus Stream?

Tanpa stream, aplikasi cenderung membaca dan memproses data secara bulk. Hal ini menimbulkan masalah:

• Konsumsi memori tinggi: Membaca file besar sekaligus bisa menyebabkan OOM.
• Lambat merespons: Menunggu semua data selesai diproses sebelum lanjut.
• Kurang scalable: Pendekatan bulk sulit diskalakan.

Dengan stream, kita hanya mengambil, memproses, dan mengirim data sepotong demi sepotong, sehingga:

• Efisien memori
• Responsif lebih cepat
• Scalable secara alami

Anatomy Sebuah Stream

Secara umum, stream memiliki beberapa komponen utama:

Diagram Flow

flowchart LR
    A[Source] --> B[Buffer]
    B --> C[Operator]
    C --> D[Sink / Consumer]

Studi Kasus: Baca File Besar dengan Stream

package main

import (
    "bufio"
    "fmt"
    "os"
)

func main() {
    file, err := os.Open("bigdata.txt")
    if err != nil {
        panic(err)
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        line := scanner.Text()
        fmt.Println("Line:", line)
    }

    if err := scanner.Err(); err != nil {
        fmt.Println("Error:", err)
    }
}

• File dibaca baris demi baris.
• Cocok untuk file besar karena hanya satu baris dimuat ke memori pada satu waktu.

Buffering dan Throttling

Buffering

Gunakan bufio.Reader untuk buffering manual:

reader := bufio.NewReader(file)
line, _ := reader.ReadString('\n')

Throttling

Gunakan time.Sleep untuk mengatur kecepatan konsumsi:

for scanner.Scan() {
    fmt.Println(scanner.Text())
    time.Sleep(100 * time.Millisecond) // simulasi throttle
}

Backpressure di Go

Go tidak memiliki mekanisme backpressure built-in di channel, tapi kamu bisa mengontrolnya lewat kapasitas buffer dan blocking behavior.

func producer(ch chan<- int) {
    for i := 0; i < 10; i++ {
        fmt.Println("Producing", i)
        ch <- i
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for data := range ch {
        fmt.Println("Consuming", data)
        time.Sleep(200 * time.Millisecond) // simulasi proses lambat
    }
}

func main() {
    ch := make(chan int, 2)
    go producer(ch)
    consumer(ch)
}

Dengan buffer kecil dan proses lambat, Go akan otomatis menunda pengiriman data dari producer (backpressure alami).

Perbandingan Bulk vs Stream

Simulasi Visual Backpressure

flowchart RL
    P(Producer) -->|Data| CH[Channel Buffer]
    CH -->|Jika tidak penuh| C(Consumer)
    C -->|Lambat memproses| Delay
    CH -->|Jika penuh| P[Producer Stalled]

Kesimpulan

Stream adalah alat penting dalam pengembangan aplikasi modern yang efisien dan scalable. Dengan memecah data menjadi bagian kecil, kita bisa mengontrol beban memori, meningkatkan respon sistem, dan menghindari bottleneck.

Go menyediakan alat seperti channel, goroutine, bufio, dan blocking operation yang cukup efektif untuk membangun sistem berbasis stream, terutama untuk I/O dan pipeline pemrosesan data.

Jika kamu ingin lanjut ke penerapan stream dalam konteks HTTP/gRPC, Kafka, atau log collector, beri tahu saja. Kita bisa lanjutkan artikelnya!