แชนเนล mpsc

Channel คืออะไร

สร้าง Channel

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send(42).unwrap();
    });
    let received = rx.recv().unwrap();
    println!("Got: {}", received);
}

send และ recv

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let msg = String::from("hello from thread");
        tx.send(msg).unwrap();
    });
    let text = rx.recv().unwrap();
    println!("{}", text);
}

สิทธิ์ความเป็นเจ้าของเคลื่อนผ่าน Channel

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let data = vec![1, 2, 3];
        tx.send(data).unwrap();
        // data is moved; cannot use it here
    });
    let v = rx.recv().unwrap();
    println!("Sum: {}", v.iter().sum::<i32>());
}

วนซ้ำผ่าน Receiver

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).unwrap();
        }
    });
    for received in rx {
        println!("Received: {}", received);
    }
}

ผู้ผลิตหลายรายด้วย clone

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();
    thread::spawn(move || { tx.send("from A").unwrap(); });
    thread::spawn(move || { tx2.send("from B").unwrap(); });
    for msg in rx {
        println!("{}", msg);
    }
}

พฤติกรรมเมื่อ Channel ปิด

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let handle = thread::spawn(move || {
        for i in 0..3 {
            tx.send(i * 10).unwrap();
        }
        // tx dropped here, closing the channel
    });
    handle.join().unwrap();
    let total: i32 = rx.iter().sum();
    println!("Total: {}", total);
}

try_recv สำหรับการอ่านแบบไม่บล็อก

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(50));
        tx.send("ready").unwrap();
    });
    loop {
        match rx.try_recv() {
            Ok(msg) => { println!("{}", msg); break; }
            Err(_) => println!("waiting..."),
        }
        thread::sleep(Duration::from_millis(20));
    }
}

ส่งชนิดข้อมูลแบบกำหนดเอง

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

enum Job {
    Print(String),
    Add(i32, i32),
}

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send(Job::Print(String::from("hi"))).unwrap();
        tx.send(Job::Add(2, 3)).unwrap();
    });
    for job in rx {
        match job {
            Job::Print(s) => println!("print: {}", s),
            Job::Add(a, b) => println!("add: {}", a + b),
        }
    }
}

sync_channel และแรงต้านกลับ

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

fn main() {
    let (tx, rx) = mpsc::sync_channel(2);
    thread::spawn(move || {
        for i in 1..=4 {
            tx.send(i).unwrap();
            println!("sent {}", i);
        }
    });
    for v in rx {
        println!("got {}", v);
    }
}

รูปแบบ Worker อย่างง่าย

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

fn main() {
    let (job_tx, job_rx) = mpsc::channel();
    let (res_tx, res_rx) = mpsc::channel();
    thread::spawn(move || {
        for n in job_rx {
            res_tx.send(n * n).unwrap();
        }
    });
    for n in 1..=4 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx);
    for r in res_rx {
        println!("square: {}", r);
    }
}

ตรวจสอบความเข้าใจอย่างรวดเร็ว

ทบทวน