L10-concurrency-1.org

CIS198 Lecture 10: 并发 I

并发

1. 并行 parallel vs 并发 concurrency
2. 分时
3. 竞争
4. 哲学家问题

线程

创建线程

1. 线程 std::thread

use std::thread;

thread::spawn(|| {
    println!("hello");
});

hello

Join

use std::thread;
let handle = thread::spawn(|| "hello");
println!("{:?}", handle.join());

Ok("hello")

Thread

let handle = thread::spawn(|| {
    println!("child thread = {:?}", thread::current());
    thread::park();
    println!("Good morning");
});
println!("parent thread = {:?}", thread::current());

println!("Good night");
handle.thread().unpark();

// join
handle.join().unwrap();

多线程，主线程 sleep

use std::{thread, time::Duration};

fn main() {
    for i in 0..10 {
        thread::spawn(move || {
            println!("aaa i = {}", i);
        });
    }

    thread::sleep(Duration::from_millis(1000));
}

Send & Sync

1. Send trait 允许安全地在线程间传递所有权。
   • 即可以跨线程移动数据
   • 除了 Rc<T> 等极少数的类型，几乎所有的 Rust 类型都实现了 Send
2. Sync trait 允许安全地在线程间传递不可变借用。
   • 即可以跨线程共享数据
   • 类型 Mutex<T> 是 Sync 的，可以被多个线程共享访问
   • 类型 RefCell<T> 不满足 Sync 约束

     use std::cell::RefCell;
     use std::thread;
     
     let a = RefCell::new(1);
     thread::spawn(|| {
         a.borrow_mut(); // error[E0277]: `RefCell<i32>` cannot be shared between threads safely
     });
             

     error: Could not compile `cargoQt5Cpp`.
             

3. 它们都是 marker trait，仅用于标记
   • Send | Sync
   • 它们都是 unsafe

共享线程状态

1. Arc 线程间共享
2. Mutex 线程安全的内部可变性

use std::{
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

fn main() {
    let data = Arc::new(Mutex::new(vec![1, 2, 3]));

    for i in 0..3 {
        let data = data.clone();
        thread::spawn(move || {
            let mut data = data.lock().unwrap();
            data[i] += 1;
        });
    }

    thread::sleep(Duration::from_millis(1000));

    println!("data = {:?}", data.clone().lock().unwrap());
}

channel

mpsc

1. mpsc (multiple producer, single consumer)
2. 通道类型 channel | sync_channel
   • channel 异步
     • tx.send(1).unwrap()
     • rx.recv().unwrap()
   • sync_channel 同步，带缓冲
3. Sender<T> 和 Reciever<T> 中的 T 会自动推导类型

use std::{sync::mpsc::channel, thread};

fn main() {
    let (tx, rx) = channel();
    thread::spawn(move || {
        for i in 0..3 {
            tx.send(i).unwrap();
        }
    });

    for r in rx {
        println!("rx = {:?}", r);
    }
}

多生成者，单消费者

use std::{sync::mpsc::channel, thread};

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

    for i in 0..3 {
        let tx = tx.clone();
        // producer
        thread::spawn(move || {
            tx.send(i).unwrap();
        });
    }

    drop(tx);

    // consumer
    // println!("rx = {:?}", rx.recv().unwrap());
    for r in rx {
        println!("rx = {:?}", r);
    }
}

mpmc

1. flume
   • A blazingly fast multi-producer, multi-consumer channel.

use std::thread;

fn main() {
    println!("Hello, world!");

    let (tx, rx) = flume::unbounded();

    thread::spawn(move || {
        (0..10).for_each(|i| {
            tx.send(i).unwrap();
        })
    });

    let received: u32 = rx.iter().sum();

    assert_eq!((0..10).sum::<u32>(), received);
}

error: Could not compile `cargoG5nBYA`.