Channel di Rust: Mengirim Pesan Antar-Thread

Di halaman Konkurensi, thread berbagi data dengan cara menguncinya memakai Arc<Mutex<T>>. Cara itu berhasil, tapi ada cara lain yang sering lebih sederhana: alih-alih berbagi data, thread saling mengirim pesan.

Bayangkan channel (saluran) seperti pipa dengan dua ujung. Satu thread memasukkan nilai ke ujung sender (pengirim). Thread lain mengambilnya dari ujung receiver (penerima), dengan urutan yang sama. Tidak ada yang berbagi apa-apa: setiap nilai berpindah dari satu thread ke thread lain. Komunitas Go punya pepatah yang juga dipinjam Rust di sini: “Jangan berkomunikasi dengan berbagi memori; sebaliknya, berbagilah memori dengan berkomunikasi.”

Channel pertamamu

Channel di standard library ada di std::sync::mpsc. Namanya singkatan dari multiple producer, single consumer: banyak pengirim, satu penerima.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("halo dari worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main menerima: {received}");
}

Output:

main menerima: halo dari worker

▶ Coba sendiri

Langkah demi langkah:

• mpsc::channel() mengembalikan sepasang nilai: sebuah Sender dan sebuah Receiver. Menurut kebiasaan, keduanya dinamai tx (transmit, mengirim) dan rx (receive, menerima).
• move menyerahkan tx ke thread baru, sama seperti kamu memindahkan data ke thread di halaman Konkurensi.
• tx.send(value) memasukkan nilai ke channel. Hasilnya berupa Result, karena pengiriman gagal kalau receiver-nya sudah tidak ada.
• rx.recv() menunggu sampai ada nilai yang datang. Ia hanya mengembalikan Err kalau semua sender sudah hilang dan channel-nya kosong.

Perhatikan bahwa kita tidak butuh join(). recv() sudah menunggu pesannya, jadi main tidak mungkin selesai terlalu cepat.

Menerima banyak nilai

Sebuah Receiver bisa dipakai sebagai iterator. Perulangan for atasnya akan terus memanggil recv() untukmu, dan perulangannya berhenti sendiri saat semua sender sudah di-drop:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx di-drop di sini, saat thread selesai
    });

    for value in rx {
        println!("dapat {value}");
    }
    println!("channel ditutup, selesai");
}

Output:

dapat 10
dapat 20
dapat 30
channel ditutup, selesai

▶ Coba sendiri

Di sini hanya ada satu sender, jadi nilai-nilainya datang persis sesuai urutan pengirimannya. Aturan “perulangan berhenti saat semua sender hilang” ini penting. Ingat baik-baik untuk bagian berikutnya.

Banyak producer: clone sender-nya

Supaya beberapa thread bisa mengirim, beri masing-masing salinan sender sendiri dengan tx.clone(). Semua salinan mengisi channel yang sama:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("worker {id} melapor")).unwrap();
        });
    }
    drop(tx); // drop yang asli, kalau tidak perulangan di bawah tidak akan selesai

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Output:

worker 1 melapor
worker 2 melapor
worker 3 melapor

▶ Coba sendiri

Dua hal yang perlu diperhatikan:

• drop(tx) itu penting. Setiap thread men-drop salinannya saat selesai, tapi main masih memegang yang asli. Selama masih ada satu sender, receiver akan terus menunggu, dan program akan macet selamanya. Men-drop-nya sama dengan berkata “main tidak akan mengirim apa-apa”.
• Thread selesai dalam urutan acak. Pesan bisa datang sebagai worker 2, worker 1, worker 3. Kita mengumpulkannya lalu menyortirnya supaya output yang dicetak selalu sama. Hapus sort() lalu jalankan beberapa kali untuk melihat urutannya berubah.

Tidak menunggu selamanya: try_recv dan recv_timeout

recv() memblokir sampai ada yang datang. Kadang kamu tidak mau menunggu. Ada dua “saudaranya” yang bisa membantu:

• try_recv() langsung kembali: berisi nilai, Empty (belum ada apa-apa), atau Disconnected (semua sender sudah hilang).
• recv_timeout(duration) menunggu, tapi hanya sampai batas waktu tertentu. Kalau tidak ada yang datang tepat waktu, kamu mendapat Timeout.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    // Belum ada yang dikirim: try_recv langsung kembali.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Sudah ada nilai yang menunggu, jadi ini langsung mengembalikannya.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // Sender-nya sudah hilang dan channel-nya kosong.
    println!("{:?}", rx.try_recv());
}

Output:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Coba sendiri

Kita memanggil join() pada thread sebelum menerima, jadi nilainya sudah menunggu dan output-nya selalu sama.

try_recv() berguna di perulangan yang punya pekerjaan lain, seperti game loop yang memeriksa pesan baru sekali per frame. recv_timeout() cocok untuk “menyerah saja kalau worker-nya terlalu lama”.

Channel berkapasitas terbatas dengan sync_channel

channel() biasa tidak punya batas ukuran. Kalau sender jauh lebih cepat daripada receiver, pesan menumpuk di memori. mpsc::sync_channel(n) membuat channel yang paling banyak menampung n pesan. Saat penuh, send() menunggu sampai receiver mengambil satu pesan. Ini disebut backpressure (tekanan balik): konsumen yang lambat otomatis memperlambat produsen yang cepat.

use std::sync::mpsc;
use std::thread;

fn main() {
    // Hanya muat 2 pesan pada satu waktu.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // memblokir selama buffer penuh
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("total = {total}");
}

Output:

total = 15

▶ Coba sendiri

Producer tidak akan pernah bisa lebih dari dua angka di depan main. Dengan sync_channel(0), tidak ada buffer sama sekali: setiap send() menunggu sampai receiver siap mengambil nilainya, seperti menyerahkan paket langsung dari tangan ke tangan.

Worker pool sederhana

Sekarang kita gabungkan semuanya. Worker pool adalah beberapa thread yang mengambil pekerjaan (job) dari antrean bersama lalu mengirim balik hasilnya. Ia memakai dua channel: satu untuk job, satu untuk hasil.

Ada satu kendala: Receiver tidak bisa di-clone (ia konsumen tunggal). Jadi para worker berbagi receiver job lewat Arc<Mutex<_>>, dan bergantian mengambil darinya:

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Receiver tidak bisa di-clone, jadi para worker berbagi satu lewat Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // tidak ada job lagi
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // memberi tahu para worker bahwa pekerjaan sudah habis

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} kuadrat adalah {square}");
    }
}

Output:

1 kuadrat adalah 1
2 kuadrat adalah 4
3 kuadrat adalah 9
4 kuadrat adalah 16
5 kuadrat adalah 25
6 kuadrat adalah 36

▶ Coba sendiri

Begini cara program ini berhenti dengan rapi:

1. drop(job_tx) menutup channel job. recv() di setiap worker lalu mengembalikan Err, dan worker keluar dari perulangannya.
2. Saat worker selesai, salinan result_tx miliknya di-drop. Yang asli sudah kita drop sebelumnya, jadi begitu ketiga worker selesai, channel hasil tertutup.
3. result_rx.iter() lalu berhenti, dan kita menyortir hasilnya karena worker selesai dalam urutan acak.

Satu detail kecil: job_rx.lock().unwrap().recv() sengaja ditulis dalam satu pernyataan let. Kuncinya dilepas di akhir pernyataan itu, jadi worker lain bisa mengambil job berikutnya selagi worker ini masih menghitung kuadrat angkanya.

Channel vs Arc<Mutex<T>>

Keduanya aman. Hanya saja masing-masing cocok untuk bentuk masalah yang berbeda:

Pakai channel kalau…	Pakai Arc<Mutex<T>> kalau…
Pekerjaan mengalir satu arah: producer membuat sesuatu, consumer mengolahnya	Banyak thread perlu membaca dan mengubah nilai yang sama (penghitung, cache)
Kamu butuh antrean job, event, atau baris log	Datanya besar dan menyalin atau memindahkannya ke sana kemari akan boros
Setiap nilai hanya punya satu pemilik pada satu waktu	Thread perlu melihat keadaan terkini, bukan aliran pembaruan

Patokan yang bagus: kalau programmu bisa digambarkan sebagai “thread-thread ini memberi tahu thread itu sesuatu”, pakai channel. Kalau “thread-thread ini sama-sama mengubah benda ini”, pakai Mutex. Banyak program sungguhan memakai keduanya, seperti worker pool kita.

Kesalahan umum

Memakai nilai setelah mengirimnya. send() mengambil ownership (kepemilikan), sama seperti mengirim nilai ke sebuah fungsi. Setelah dikirim, nilai itu milik receiver:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("hai");
        tx.send(message).unwrap();
        println!("terkirim {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:29
   |
 8 |         let message = String::from("hai");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("terkirim {message}");
   |                             ^^^^^^^ value borrowed here after move

Pesan error compiler memang selalu dalam bahasa Inggris. Intinya: message sudah dipindahkan (moved) oleh send, lalu dipinjam lagi di println!. Inilah cara Rust melindungimu: kalau pengirim masih bisa mengubah message, dua thread akan menyentuh data yang sama. Cetak sebelum mengirim, atau kirim .clone()-nya.

Memindahkan sender ke dalam perulangan yang membuat thread. Thread pertama mengambil tx, jadi tidak ada yang tersisa untuk thread kedua:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

Artinya: tx sudah dipindahkan ke closure pada putaran perulangan sebelumnya. Perbaikannya adalah let tx = tx.clone(); di awal perulangan, seperti di contoh banyak producer.

Program yang tidak pernah selesai. Kasus ini tidak memunculkan error compiler sama sekali, dan justru itu yang membuatnya licik. Kalau kamu melakukan perulangan atas receiver (for x in rx) sementara masih ada sender yang hidup, sering kali tx asli di main, perulangannya akan menunggu selamanya. Kalau programmu “tiba-tiba macet”, periksa apakah setiap sender sudah di-drop.

Langkah berikutnya

std::sync::mpsc sudah mencakup banyak hal, tapi kamu akan bertemu dua alternatif populer:

• crossbeam-channel (bahasa Inggris) menambahkan channel dengan banyak consumer (receiver yang bisa di-clone) dan makro select! untuk menunggu beberapa channel sekaligus.
• tokio::sync::mpsc adalah channel untuk kode async. Kalau kamu memakai async/.await (lihat Async/Await), pakai channel ini alih-alih yang dari std, karena recv().await-nya tidak memblokir seluruh thread.

Ringkasan

• mpsc::channel() memberimu pasangan (Sender, Receiver). send() memasukkan nilai, recv() menunggu nilai berikutnya.
• Clone sender-nya untuk punya banyak producer. Receiver-nya selalu hanya satu.
• Perulangan atas receiver berhenti saat semua sender sudah di-drop, jadi jangan lupa drop() yang asli.
• try_recv() dan recv_timeout() memungkinkanmu memeriksa tanpa menunggu selamanya. sync_channel(n) membatasi antrean sampai n pesan.
• Pesan dari beberapa thread datang dalam urutan acak: sortir atau jumlahkan kalau kamu butuh output yang stabil.
• Pakai channel saat data mengalir dari thread ke thread. Pakai Arc<Mutex<T>> saat thread-thread berbagi dan mengubah satu nilai.

Kuis singkat

Kuis singkat

1. "mpsc" itu singkatan dari apa?

Many processes, shared cache Message passing, safe concurrency Multiple producer, single consumer Mutex protected shared channel

2. Bagaimana cara membuat tiga thread bisa mengirim ke channel yang sama?

Beri setiap thread sebuah tx.clone() Bungkus receiver-nya dengan Rc Panggil mpsc::channel() tiga kali lalu gabungkan Tidak bisa; channel hanya punya satu sender

3. Perulangan for value in rx tidak pernah selesai. Apa penyebab yang paling mungkin?

Channel-nya penuh Ada sender (sering kali tx asli di main) yang tidak pernah di-drop recv() dipanggil terlalu cepat Nilainya bukan Copy

4. Apa yang dilakukan send() saat sync_channel(2) sudah berisi 2 pesan?

Membuang pesan paling lama Langsung mengembalikan error Panic Menunggu sampai receiver mengambil satu pesan

Skor: 0 / 4

Bacaan lebih lanjut

• Using Message Passing to Transfer Data Between Threads — The Rust Book (bahasa Inggris): pengenalan resmi tentang channel, dengan lebih banyak contoh.
• std::sync::mpsc — dokumentasi standard library (bahasa Inggris): semua method milik Sender, SyncSender dan Receiver.
• Channels — Rust by Example (bahasa Inggris): contoh singkat yang bisa dijalankan, dengan beberapa thread producer.
• From Single-Threaded to Multithreaded Server — The Rust Book (bahasa Inggris): membangun thread pool lengkap dengan channel dan Arc<Mutex<Receiver>>, langkah demi langkah.
• tokio::sync::mpsc — docs.rs (bahasa Inggris): channel async yang akan kamu pakai bersama Tokio, dan bedanya dengan yang dari std.