Satu producer yang mengirim semua pesan hanyalah pemindahan loop for dari satu thread ke thread lain — beban pengiriman tetap terpusat pada satu alur. Kekuatan sesungguhnya dari nama mpsc justru pada huruf pertamanya: multi-producer. Bayangkan tiga cabang toko yang sama-sama melaporkan penjualan ke satu kantor pusat: tiap cabang punya ritme kerja sendiri, dan pusat cukup duduk di satu channel menerima laporan dari semua arah. Rust mendukung arsitektur ini dengan satu pernyataan sederhana — tx.clone() — yang menghadirkan jebakan kecil namun penting: channel hanya dianggap tertutup setelah seluruh salinan transmitter mati. Chapter ini membedah pola lengkapnya beserta perilaku antrean saat banyak pengirim berebut.

Mengkloning Transmitter #

Tipe Sender<T> mengimplementasikan Clone. Hasil tx.clone() bukan channel baru, melainkan pegangan tambahan menuju antrean yang sama — dua transmitter, satu jalur. Karena setiap pegangan harus dipindahkan ke thread-nya masing-masing lewat closure move, pola yang lahir selalu berbentuk tiga langkah: kloning tx di dalam loop, pindahkan kloningan itu ke thread baru, lalu jatuhkan transmitter asli di akhir agar penerima tahu kapan aliran pesan berakhir.

Aturan penutupan inilah yang membuat drop(tx) menjadi wajib pada pola banyak producer. Penerima loop for item in rx hanya berhenti setelah semua sender — kloningan-kloningan di worker dan transmitter asli — keluar dari dunia. Kloningan ikut mati secara otomatis saat worker selesai; transmitter asli di tangan main tidak akan mati sendiri selama main masih hidup, sehingga harus dijatuhkan secara eksplisit.

Mari kita lihat pola lengkapnya dengan program penghitung pajak: tiga worker masing-masing menghitung pajak satu karyawan, lalu main menjumlahkan semuanya:

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

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

    let gaji_bulanan = vec![1_500_000, 1_750_000, 2_100_000];

    for gaji in gaji_bulanan {
        let tx_karyawan = tx.clone();
        thread::spawn(move || {
            let pajak = gaji / 10;
            tx_karyawan.send(pajak).unwrap();
        });
    }

    drop(tx);

    let total_pajak: i32 = rx.iter().sum();
    println!("Total pajak karyawan: Rp{}", total_pajak);
}

Output yang Dihasilkan:

text 
Total pajak karyawan: Rp535000

Mari kita bedah kode di atas:

  1. Baris 10 (let tx_karyawan = tx.clone();): setiap iterasi membuat satu pegangan transmitter baru menuju antrean yang sama. Variabel tx_karyawan akan dipindahkan ke worker — sementara tx tetap dipegang main untuk iterasi berikutnya.
  2. Baris 11 (thread::spawn(move || {): closure move mengambil dua nilai sekaligus: tx_karyawan (pegangan kirim) dan gaji (angka bertipe i32, tersalin karena Copy). Worker sekarang mandiri penuh — tuntutan 'static terpenuhi.
  3. Baris 12 (let pajak = gaji / 10;): perhitungan dijalankan paralel di tiga thread yang berbeda. Tidak ada data yang dibagikan, sehingga tidak ada pula yang perlu dikunci.
  4. Baris 17 (drop(tx);): baris paling menentukan. Setelah loop selesai, tx asli dijatuhkan; tiga pegangan di worker menyusul mati saat worker rampung. Setelah transmitter terakhir jatuh, channel resmi tertutup.
  5. Baris 19 (let total_pajak: i32 = rx.iter().sum();): iter() menghasilkan iterator di atas rx yang menunggu pesan satu per satu; sum() mengumpulkan ketiganya lalu berhenti saat channel tertutup. Penjumlahan bersifat komutatif — berapa pun urutan kedatangan pesan, total selalu 535000, dan output program ini pasti identik pada setiap eksekusi.

Bentuk alurnya dapat digambarkan sebagai pola fan-in menuju satu penerima:

text 
worker gaji 1.5 juta ──┐
worker gaji 1.75 juta ──┼──> antrean channel ──> rx di main
worker gaji 2.1 juta  ──┘

Diagram di atas menegaskan satu hal: jalur dan antreannya tunggal — yang banyak hanyalah pintu masuknya. Ketiadaan dua antrean berarti tidak ada satu pun pesan yang hilang di jalan, dan tidak ada satu pun penerima yang berebut.

Antrean Bersama dari Banyak Thread #

Contoh pajak tadi menyembunyikan satu fakta karena sum tidak peduli urutan. Saat urutan kedatangan diperhatikan, sifat non-deterministik pola ini perlu kita tatap langsung. Program berikut menjalankan tiga cabang yang masing-masing mengirim beberapa transaksi; main mencetak setiap transaksi begitu tiba:

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

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

    let cabang = vec![
        ("Jakarta", vec![12000, 8000]),
        ("Bandung", vec![15000]),
        ("Surabaya", vec![9000, 17500]),
    ];

    for (nama, pesanan) in cabang {
        let tx_cabang = tx.clone();
        thread::spawn(move || {
            for nilai in pesanan {
                tx_cabang.send((nama, nilai)).unwrap();
            }
        });
    }

    drop(tx);

    let mut total = 0;
    for (nama, nilai) in rx {
        println!("{}: Rp{}", nama, nilai);
        total += nilai;
    }
    println!("Total: Rp{}", total);
}

Output yang Dihasilkan:

text 
Surabaya: Rp9000
Jakarta: Rp12000
Surabaya: Rp17500
Bandung: Rp15000
Jakarta: Rp8000
Total: Rp61500

Mari kita bedah kode di atas:

  1. Baris 7-11 (let cabang = vec![...]): data dibentuk sebagai daftar pasangan — tiap cabang membawa nama dan daftar transaksinya. Tuple (&str, Vec<i32>) berpindah utuh ke worker lewat move.
  2. Baris 13 (for (nama, pesanan) in cabang {): destructuring mengurai tiap pasangan menjadi nama dan pesanan sebelum dikirim ke thread.
  3. Baris 17 (tx_cabang.send((nama, nilai)).unwrap();): pesan yang dikirim berupa tuple (nama, nilai). nama bertipe &str yang bersifat Copy, sehingga bisa dikirim berulang kali tanpa mati; nilai bertipe i32 juga Copy.
  4. Baris 22 (drop(tx);): kunci penutupan yang sama seperti sebelumnya — tanpa baris ini, loop for ... in rx di Baris 25 akan menunggu selamanya.
  5. Baris 25-28 (for (nama, nilai) in rx { ... }): penerima mencetak tiap transaksi saat tiba, lalu menambahkannya ke total.

Output di atas adalah hasil satu eksekusi nyata, dan baris urutan antar cabangnya dapat berbeda di setiap eksekusi. Tiga thread berjalan di irisan waktu penjadwal OS; siapa yang lebih dulu menempuh jalur send, pesannya lebih dulu mendarat di antrean. Dua jaminan tetap berdiri: pesan dari satu cabang yang sama selalu tiba berurutan (Surabaya 9000 selalu mendahului Surabaya 17500, karena pengirimannya dari satu thread yang sama melalui antrean FIFO), dan total di baris terakhir selalu 61500 karena penjumlahan tidak peduli urutan. Pola umumnya layak diingat: hasil akhir dari agregasi komutatif itu deterministik; urutan kedatangan pesan tidak.

Ringkasan #

Sender bisa dikloning: setiap tx.clone() adalah pintu masuk baru menuju satu antrean yang sama, dan channel hanya tertutup setelah transmitter asli beserta seluruh kloningannya dijatuhkan. Pola tiga langkahnya — kloning di loop, move ke worker, drop(tx) di akhir — membentuk arsitektur fan-in yang menghubungkan banyak producer ke satu consumer. Urutan kedatangan pesan lintas producer tidak menentu karena penjadwal OS, tetapi urutan dari satu producer terjaga oleh FIFO dan hasil agregasi komutatif tetap deterministik. Ketika data yang harus dibagi bukan pesan yang mengalir, melainkan satu nilai yang semua thread perlu baca-tulis di tempat, kita butuh perangkat lain: Mutex — Chapter berikutnya.