Coroutine Channels

Sampai titik ini, kita memakai async untuk mengembalikan satu nilai dari coroutine. Bagaimana jika kita ingin mengirim banyak nilai seiring waktu — mis. event dari sensor, baris dari database, atau pesan dari socket? Di sinilah Channel<T> masuk. Channel adalah hot stream — produsen dan konsumen hidup secara independen, dan nilai mengalir di antara mereka. Bab ini membahas dasar Channel, perbedaan mode buffer, dan builder produce untuk membuat producer coroutine yang idiomatik.

Konsep Channel #

Channel<T> adalah antrean konkuren yang menghubungkan sender dan receiver. Karakteristik utama:

Empat mode kapasitas:

Tiga Pola Channel #

Kode berikut mendemonstrasikan tiga pola: rendezvous, buffered, dan builder produce:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runBlocking {
    // 1. Channel rendezvous (default): kapasitas 0, send menunggu receive
    println("=== 1. Channel rendezvous ===")
    val ch1 = Channel<String>()
    launch {
        repeat(3) { i ->
            println("[producer] kirim $i")
            ch1.send("pesan-$i")
        }
        ch1.close()
    }
    for (msg in ch1) {
        println("[consumer] terima $msg")
    }

    // 2. Channel buffered: kapasitas 2
    println()
    println("=== 2. Channel buffered (kapasitas 2) ===")
    val ch2 = Channel<Int>(2)
    launch {
        repeat(4) { i ->
            println("[producer] kirim $i")
            ch2.send(i)
        }
        ch2.close()
    }
    for (msg in ch2) {
        println("[consumer] terima $msg, antrian tersisa sebelum terima: -")
    }

    // 3. Channel produce/consume
    println()
    println("=== 3. produce builder ===")
    val producer = produce {
        repeat(3) { i ->
            send("data-$i")
        }
    }
    repeat(3) {
        println("[consumer] terima: ${producer.receive()}")
    }
    producer.cancel()
}

Output yang Dihasilkan:

=== 1. Channel rendezvous ===
[producer] kirim 0
[producer] kirim 1
[consumer] terima pesan-0
[consumer] terima pesan-1
[producer] kirim 2
[consumer] terima pesan-2

=== 2. Channel buffered (kapasitas 2) ===
[producer] kirim 0
[producer] kirim 1
[producer] kirim 2
[producer] kirim 3
[consumer] terima 0, antrian tersisa sebelum terima: -
[consumer] terima 1, antrian tersisa sebelum terima: -
[consumer] terima 2, antrian tersisa sebelum terima: -
[consumer] terima 3, antrian tersisa sebelum terima: -

=== 3. produce builder ===
[consumer] terima: data-0
[consumer] terima: data-1
[consumer] terima: data-2

Mari kita telaah baris-baris kode di atas:

1. Baris 8-19 (rendezvous): Channel<String>() tanpa argumen membuat channel dengan kapasitas 0. Setiap send menangguhkan sampai receive siap. Producer dan consumer harus bertemu — itulah asal nama “rendezvous”. Output menunjukkan pola: producer mencetak “kirim 0”, lalu send(0) menangguhkan. Consumer menerima, lalu producer melanjutkan ke “kirim 1” — pola ini berulang.
2. Baris 14 (ch1.close()): menandai bahwa tidak ada lagi nilai yang akan dikirim. Channel for loop (baris 17) berhenti secara natural saat channel ditutup dan kosong. Tanpa close(), for loop akan menangguhkan selamanya menunggu nilai berikutnya.
3. Baris 23-33 (buffered kapasitas 2): Channel<Int>(2) membuat buffer 2 slot. Producer dapat mengirim hingga 2 nilai sebelum harus menunggu consumer. Output menunjukkan producer mengirim 0, 1 (buffer terisi), lalu 2 (menangguhkan karena buffer penuh). Consumer membaca 0 (slot bebas), producer melanjutkan ke 3, dst.
4. Baris 38-43 (produce { ... }): builder produce membuat coroutine producer yang mengembalikan ReceiveChannel<T>. Channel ini hanya bisa diterima (receive) — tidak bisa dikirim dari luar. Builder ini idiomatik untuk pola “satu producer, satu consumer”.
5. Baris 44-46 (repeat(3) { producer.receive() }): menerima 3 nilai dari producer. Setelah selesai, producer.cancel() membersihkan producer coroutine. Tanpa cancel(), producer akan terus hidup sampai scope berakhir.

Send dan Receive #

Dua operasi utama channel:

• send(value): suspend function. Menangguhkan jika buffer penuh (atau untuk rendezvous, jika tidak ada receiver).
• receive(): suspend function. Menangguhkan jika channel kosong. Mengembalikan nilai berikutnya atau melempar ClosedReceiveChannelException jika channel ditutup dan kosong.

val ch = Channel<Int>()
launch {
    ch.send(1)
    ch.send(2)
    ch.close()
}
println(ch.receive())  // 1
println(ch.receive())  // 2
// ch.receive() akan lempar karena ditutup

Iterasi dan Penutupan #

Channel dapat diiterasi dengan for loop:

for (item in channel) {
    process(item)
}
// iterasi berhenti saat channel ditutup

Untuk menutup channel dari sisi producer, panggil close(). Untuk membatalkan consumer, gunakan cancel() (membatalkan channel) — producer akan menerima ClosedSendChannelException saat mencoba send berikutnya.

Kasus Penggunaan Channel #

Channel cocok untuk:

1. Pipa antar coroutine — mengalirkan data dari satu coroutine ke coroutine lain (mis. parser → transformer → writer).
2. Worker pool — mendistribusikan tugas ke banyak worker coroutine.
3. Event fan-out — satu producer mengirim event ke banyak consumer (perlu broadcast, bukan channel tunggal).
4. Throttling — membatasi kecepatan produksi dengan buffer kapasitas kecil.

suspend fun proses(items: List<Work>): List<Result> = coroutineScope {
    val input = Channel<Work>()
    val output = Channel<Result>()
    // 4 worker
    repeat(4) {
        launch {
            for (w in input) {
                output.send(processOne(w))
            }
        }
    }
    // producer
    launch {
        for (item in items) input.send(item)
        input.close()
    }
    // collector
    val hasil = mutableListOf<Result>()
    repeat(items.size) { hasil.add(output.receive()) }
    hasil
}

Referensi Lanjutan #

• Channels — panduan resmi Channel<T> untuk komunikasi antar-coroutine
• kotlinx.coroutines core — API reference kotlinx.coroutines.channels

Selanjutnya #

Channel cocok untuk komunikasi satu-ke-satu. Untuk multicast (satu-ke-banyak), Kotlin menyediakan SharedFlow. Bab Shared Flows akan memperkenalkan konsep hot multicast stream dan bagaimana SharedFlow memungkinkan banyak collector berbagi emisi yang sama.