Channels

การส่งข้อความด้วย `std::sync::mpsc`

standard library ของ Rust มี channel แบบ multi-producer, single-consumer (mpsc) สำหรับส่งค่าระหว่าง thread channel มีสองปลาย:

tx — transmitter (ผู้ส่ง) คุณ clone ตัวนี้เพื่อสร้าง producer หลายตัวได้
rx — receiver (ผู้รับ) มี receiver เพียงตัวเดียวต่อ channel

use std::sync::mpsc;

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

    tx.send(42).unwrap();

    let value = rx.recv().unwrap(); // บล็อกจนกว่าค่าจะมาถึง
    println!("Received: {}", value);
}

mpsc::channel() ส่งคืนคู่ (Sender<T>, Receiver<T>) ประเภท T ถูก infer จาก send call แรก

การส่งค่าข้าม Thread

transmitter เป็น Send ดังนั้นคุณย้ายเข้าไปใน thread closure ได้ เมื่อ tx.send(value) ถูกเรียก ค่านั้นจะถูก ย้าย เข้าไปใน channel — thread ผู้ส่งไม่ได้เป็นเจ้าของค่านั้นอีกต่อไป

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

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

    thread::spawn(move || {
        let s = String::from("hello from thread");
        tx.send(s).unwrap();
        // println!("{}", s); // จะไม่คอมไพล์ — s ถูกย้ายเข้า channel แล้ว
    });

    let received = rx.recv().unwrap();
    println!("Main got: {}", received);
}

Multiple Producers

Clone tx เพื่อสร้าง sender หลายตัว แต่ละ clone มี handle ของตัวเองไปยัง channel เดียวกัน receiver เห็นข้อความตามลำดับที่มาถึง (non-deterministic ข้าม thread) ดังนั้นให้รวบรวมและ sort ก่อนพิมพ์

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let mut handles = Vec::new();

    for i in 0..3 {
        let tx = tx.clone(); // แต่ละ thread ได้รับ clone ของ sender
        handles.push(thread::spawn(move || {
            tx.send(format!("message {}", i)).unwrap();
        }));
    }
    drop(tx); // drop tx ตัวเดิมเพื่อให้ channel ปิดเมื่อ clone ทั้งหมดหายไป

    for h in handles {
        h.join().unwrap();
    }

    let mut msgs: Vec<String> = rx.iter().collect(); // รวบรวมจนกว่า channel จะปิด
    msgs.sort();
    for m in &msgs {
        println!("{}", m);
    }
}

rx.iter() ให้ค่าจนกว่า Sender handle ทั้งหมดจะถูก drop การ drop tx ตัวเดิม (หลัง clone) จำเป็นเพื่อให้ iterator รู้ว่าจะหยุดเมื่อใด

Fan-Out: การกระจายงาน

producer คนเดียวสามารถส่ง work item ไปยัง pool ของ thread แต่ละ worker รับ item ผ่าน channel ของตัวเอง

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

fn main() {
    let (main_tx, main_rx) = mpsc::channel::<String>();
    let mut handles = Vec::new();

    for worker_id in 0..3 {
        let (work_tx, work_rx) = mpsc::channel::<i32>();
        let main_tx = main_tx.clone();

        handles.push(thread::spawn(move || {
            let item = work_rx.recv().unwrap();
            main_tx.send(format!("worker {} processed {}", worker_id, item * 2)).unwrap();
        }));

        work_tx.send(worker_id * 10).unwrap();
    }
    drop(main_tx);

    for h in handles {
        h.join().unwrap();
    }

    let mut results: Vec<String> = main_rx.iter().collect();
    results.sort();
    for r in &results {
        println!("{}", r);
    }
}

ตัวอย่าง Playground แบบสมบูรณ์

Open in Rust Playground

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let mut handles = Vec::new();

    for i in 0..5 {
        let tx = tx.clone();
        let handle = thread::spawn(move || {
            let msg = format!("result from thread {}", i);
            tx.send(msg).unwrap();
        });
        handles.push(handle);
    }
    drop(tx);

    for h in handles {
        h.join().unwrap();
    }

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