Channel: apa itu, jenis saluran dan korutin di Kotlin

Apa itu Channel di Kotlin?

Channel — secara konseptual mirip dengan BlockingQueue dari Java, tetapi dengan fungsi suspend send() dan receive() sebagai pengganti put() dan take() yang memblokir. Pengembang Kotlin menggunakan Channel untuk mengatur pertukaran data antar korutin tanpa sinkronisasi melalui memori bersama. Saluran menjamin pengiriman terurut — urutan pengiriman sesuai dengan urutan penerimaan.

Membuat saluran

Untuk membuat Channel, fungsi pabrik Channel<T>(capacity) dipanggil. Parameter capacity menentukan jenis saluran: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) atau angka tertentu. Tipe elemen T ditentukan oleh generik. Penutupan saluran melalui close() menandakan bahwa tidak akan ada elemen baru.

Send dan Receive

send(value) — fungsi suspend yang menangguhkan korutin pengirim jika saluran penuh. receive() — fungsi suspend yang menangguhkan penerima jika saluran kosong. Alternatif trySend() dan tryReceive() — versi non-memblokir yang mengembalikan Boolean atau null ketika operasi tidak memungkinkan. Berguna dalam konteks non-suspend.

Jenis Channel di kotlinx.coroutines

Kotlin menyediakan empat varian Channel melalui kapasitas buffer: Rendezvous (kapasitas 0), Buffered (kapasitas N), Conflated (kapasitas 1, timpa) dan Unlimited (kapasitas Int.MAX_VALUE). Setiap jenis memecahkan tugasnya sendiri, dari sinkronisasi ketat hingga buffering data massal.

Rendezvous Channel — yang paling ketat: send() memblokir sampai receive() dipanggil di korutin lain. Pada dasarnya, ini adalah titik rendezvous dua korutin. Ideal untuk handshake ketat, ketika pengirim harus menunggu penerima memproses elemen. Kehilangan data tidak mungkin — send tidak selesai sampai receive dieksekusi.

Conflated Channel — hanya menyimpan nilai terakhir yang dikirim. Jika pengirim menempatkan elemen baru sebelum penerima mengambil yang lama, yang lama dibuang. Conflated Channel berguna untuk status UI: jika pengguna dengan cepat mengubah slider, nilai antara dapat dibuang dan hanya yang terakhir diproses.

Producer-Consumer pada saluran

Pola klasik Producer-Consumer pada Channel diimplementasikan melalui korutin paralel. Producer dalam loop memanggil send(value), consumer — receive(value). Produsen dan konsumen dapat bekerja pada Dispatchers berbeda: producer pada Dispatchers.IO, consumer pada Dispatchers.Main. Channel secara otomatis menyinkronkan akses tanpa Lock dan synchronized.

Fan-out — beberapa konsumen pada satu saluran. Setiap elemen akan sampai ke tepat satu konsumen (distribusi round-robin). Fan-in — beberapa produsen menulis ke satu saluran. Korutin pengirim bersaing untuk pengiriman, tetapi urutan elemen dipertahankan. Kedua skenario tidak memerlukan sinkronisasi tambahan.

Produce — adalah builder korutin yang membuat saluran dengan penutupan otomatis. Fungsi produce { } mengembalikan ReceiveChannel — saluran read-only untuk konsumen. Di dalam builder, send() mengirim data, dan saat blok selesai atau terjadi pengecualian, saluran otomatis ditutup, mencegah kebocoran.

Select dan multipleksing

Pustaka kotlinx.coroutines menyediakan select — ekspresi yang menunggu saluran pertama selesai dari beberapa alternatif. Select memungkinkan multipleksing beberapa saluran: misalnya, menunggu data dari dua sumber dan memproses yang merespons pertama. Sintaks — select<T> { channel1.onReceive { } channel2.onReceive { } }. Ini adalah alternatif untuk operator amb di Rx.

Contoh kode Channel

Contoh pertama — Rendezvous Channel paling sederhana, di mana pengirim menunggu penerimaan:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Terkirim")
}

scope.launch {
    val msg = channel.receive()
    println("Diterima: $msg")
}

Contoh kedua — beberapa konsumen pada satu saluran (fan-out):

val channel = Channel<Int>(Channel.UNLIMITED)

scope.launch {
    for (x in 1..10) channel.send(x)
    channel.close()
}

repeat(2) { id ->
    scope.launch {
        for (msg in channel) {
            println("Konsumen #$id: $msg")
        }
    }
}

Contoh ketiga — penggunaan builder produce dengan penanganan kesalahan:

val source = produce {
    for (i in 1..5) {
        delay(200)
        send(i)
    }
}

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Kesalahan: $it") }
        .collect { println("Elemen: $it") }
}

Channel vs Flow

Channel — adalah primitif hot: data dipancarkan terlepas dari pelanggan. Flow — cold: data dihasilkan saat berlangganan. Channel mendukung banyak produsen dan konsumen dengan pengiriman terjamin setiap elemen ke satu konsumen (fan-out). Flow tidak dirancang untuk banyak produsen independen.

Channel menggunakan buffer dengan kapasitas yang dapat dikonfigurasi dan fungsi suspend send/receive untuk manajemen backpressure. Flow menggunakan mekanisme suspend collect dengan backpressure otomatis melalui korutin. Channel — alat tingkat rendah untuk skenario spesifik: konversi callback, model actor, antrian tugas dengan banyak pengirim.

Untuk skenario sehari-hari di Android (status UI, stream reaktif dari database) Google merekomendasikan Flow, bukan Channel. Channel harus digunakan ketika pertukaran data hot antar korutin dengan kontrol buffer yang presisi diperlukan, atau saat konversi antarmuka callback melalui callbackFlow, yang implementasi internalnya menggunakan Channel.

Contoh praktis penting: dalam implementasi klien WebSocket, Channel memungkinkan menulis pesan dari satu korutin dan membaca dari yang lain dengan jaminan bahwa setiap pesan akan diproses tepat satu kali. Flow tidak cocok untuk tugas ini karena bersifat cold dan tidak mendukung banyak produsen. Channel dengan kapasitas UNLIMITED memastikan bahwa pesan masuk tidak hilang selama penundaan sementara konsumen.

Manajemen siklus hidup saluran — bagian penting dari bekerja dengan Channel. Saluran harus ditutup ketika semua data telah dikirim, agar konsumen dapat menyelesaikan iterasi. Panggilan channel.close() menandakan bahwa tidak akan ada elemen baru. Konsumen dapat melakukan iterasi melalui for (item in channel) — loop berakhir secara otomatis setelah close() dan pengosongan buffer. Alternatifnya, konsumen dapat memanggil receive() dalam loop dengan penanganan ClosedReceiveChannelException.

Channel secara aktif digunakan di Android untuk mengimplementasikan EventBus tanpa ketergantungan: Channel<Event> global dengan strategi Broadcast memungkinkan mengirim acara dari titik mana pun dalam aplikasi. Tidak seperti bus berbasis LiveData, Channel tidak terikat pada lifecycle dan tidak memerlukan pengaturan ulang saat transisi antar layar. send() dari ViewModel dan receive() di Activity/Fragment melalui lifecycleScope memberikan komunikasi type-safe tanpa kelas Event. Banyak konsumen pada Channel mendistribusikan beban — setiap elemen diproses satu kali, mencegah duplikasi pemrosesan acara yang sama di pelanggan berbeda.

Dalam sistem aktor, Channel berfungsi sebagai dasar untuk mengimplementasikan mailbox — antrian pesan untuk aktor. Aktor — adalah korutin yang dalam loop membaca pesan dari Channel dan memprosesnya secara sekuensial. Pendekatan ini menjamin bahwa setiap pesan diproses dalam urutan pengiriman, tanpa race condition data. Kotlin tidak memiliki aktor bawaan sebagai tipe (tidak seperti Akka), tetapi Channel + launch adalah pengganti ringan.

Untuk pertukaran dua arah digunakan pasangan saluran: satu saluran untuk permintaan dari klien ke server, yang kedua — untuk respons dari server ke klien. Misalnya, dalam implementasi Pipe di aplikasi multithread: produsen menulis ke OutputChannel, konsumen membaca dari InputChannel. Fungsi suspend send dan receive menjamin bahwa Producer-Consumer tidak akan meluapkan tumpukan panggilan, karena korutin ditangguhkan, bukan diblokir. Channel dengan capacity BUFFERED cocok untuk sebagian besar skenario di mana kecepatan produsen dan konsumen kira-kira sama. Untuk skenario asimetris, gunakan UNLIMITED agar produsen tidak ditangguhkan saat konsumen sibuk — ini mengurangi risiko deadlock, tetapi meningkatkan konsumsi memori.

Memilih kapasitas Channel

Saat merancang arsitektur pada saluran, penting untuk mengingat capacity: pilihan kapasitas secara langsung mempengaruhi perilaku saat beban puncak. Saluran dengan kapasitas BUFFERED(N) bertindak sebagai buffer perata: jika konsumen sementara lebih lambat dari produsen, elemen menumpuk. Jika kecepatan rata-rata konsumen secara stabil lebih rendah dari produsen, buffer akan terisi dan korutin pengirim akan ditangguhkan — ini adalah backpressure otomatis yang melindungi dari kelebihan memori.

Untuk pemantauan dan debugging Channel, gunakan kotlinx-coroutines-debug: alat ini menunjukkan jumlah korutin aktif, status saluran mereka (terbuka/tertutup, jumlah elemen dalam buffer), dan tumpukan panggilan operasi send/receive yang ditangguhkan. Channel juga dapat dibungkus dalam proxy pencatat: kelas LoggingChannel<T> mendelegasikan panggilan ke Channel asli, mencatat operasi send, receive, dan close. Ini membantu mengidentifikasi kebocoran saluran ketika close() tidak dipanggil dan korutin konsumen menunggu selamanya elemen baru.