Rust语言学习笔记（13）并发

线程

// 子线程
thread::spawn(|| {
    for i in 1..10 {
        println!("hi number {} from the spawned thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
});
// 主线程
for i in 1..5 {
    println!("hi number {} from the main thread!", i);
    thread::sleep(Duration::from_millis(1));
}
// 主线程结束后子线程被中止
/*
hi number 1 from the main thread!
hi number 1 from the spawned thread!
hi number 2 from the main thread!
hi number 2 from the spawned thread!
hi number 3 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the main thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
*/
// join 连结主线程和子线程
let handle = thread::spawn(|| {
    for i in 1..10 {
        println!("hi number {} from the spawned thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
});
for i in 1..5 {
    println!("hi number {} from the main thread!", i);
    thread::sleep(Duration::from_millis(1));
}
handle.join().unwrap();
/*
hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 1 from the spawned thread!
hi number 3 from the main thread!
hi number 2 from the spawned thread!
hi number 4 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
hi number 6 from the spawned thread!
hi number 7 from the spawned thread!
hi number 8 from the spawned thread!
hi number 9 from the spawned thread!
*/
let handle = thread::spawn(|| {
    for i in 1..10 {
        println!("hi number {} from the spawned thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
});
handle.join().unwrap();
for i in 1..5 {
    println!("hi number {} from the main thread!", i);
    thread::sleep(Duration::from_millis(1));
}
/*
hi number 1 from the spawned thread!
hi number 2 from the spawned thread!
hi number 3 from the spawned thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
hi number 6 from the spawned thread!
hi number 7 from the spawned thread!
hi number 8 from the spawned thread!
hi number 9 from the spawned thread!
hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 3 from the main thread!
hi number 4 from the main thread!
*/
// 由于子线程中的闭包的生命期和主线程可能不一致
// 所以必须使用移动捕获来转移主线程中资源的所有权
let v = vec![1, 2, 3];
let handle = thread::spawn(move || {
    println!("Here's a vector: {:?}", v);
});
handle.join().unwrap();

信道（Channel）

// 创建信道
let (tx, rx) = mpsc::channel();
// 在子线程中通过信道发送值
thread::spawn(move || {
    let val = String::from("hi");
    tx.send(val).unwrap();
    // println!("val is {}", val); // error: use of moved value: `val`
});
// 在主线程中通过信道接收值
let received = rx.recv().unwrap();
println!("Got: {}", received); // Got: hi

• mpsc 的意思是 multiple producer, single consumer （多个生产者，一个消费者）
  可以想象成多条细流汇合成一条大河
• tx 是发送端（tansmitter），rx 是接收端（receiver）
• 子线程通过移动捕获获得发送端的所有权
• 发送和接收均有可能失败
  接收端被释放（drop）时发送会失败
  发送端被关闭（close）时接收会失败
• 值被发送意味着所有权被转移

// 传送多个值
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
    let vals = vec![
        String::from("hi"),
        String::from("from"),
        String::from("the"),
        String::from("thread"),
    ];
    for val in vals {
        tx.send(val).unwrap();
        thread::sleep(Duration::from_secs(1));
    }
});
for received in rx {
    println!("Got: {}", received);
}
/*
Got: hi
Got: from
Got: the
Got: thread
*/

// 多个发送者
let (tx, rx) = mpsc::channel();
let tx1 = mpsc::Sender::clone(&tx);
thread::spawn(move || {
    let vals = vec![
        String::from("hi"),
        String::from("from"),
        String::from("the"),
        String::from("thread"),
    ];
    for val in vals {
        tx1.send(val).unwrap();
        thread::sleep(Duration::from_secs(1));
    }
});
thread::spawn(move || {
    let vals = vec![
        String::from("more"),
        String::from("messages"),
        String::from("for"),
        String::from("you"),
    ];
    for val in vals {
        tx.send(val).unwrap();
        thread::sleep(Duration::from_secs(1));
    }
});
for received in rx {
    println!("Got: {}", received);
}
/*
Got: hi
Got: more
Got: from
Got: messages
Got: for
Got: the
Got: thread
Got: you
*/

Mutex<T>

Mutex 的意思是 mutual exclusion （相互排斥）

let m = Mutex::new(5);
{
    let mut num = m.lock().unwrap();
    *num = 6;
}
println!("m = {:?}", m); // m = Mutex { data: 6 }

Arc<T>

Arc 的意思是 atomically reference counted type （原子性引用计数类型）

let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
    let counter = Arc::clone(&counter);
    let handle = thread::spawn(move || {
        let mut num = counter.lock().unwrap();
        *num += 1;
    });
    handles.push(handle);
}
for handle in handles {
    handle.join().unwrap();
}
println!("Result: {}", *counter.lock().unwrap()); // Result: 10