Rust学习实践（五） -- Rust特性：并发编程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
use std::thread;
use std::time::Duration;
fn test_spwan() {
    // 获取thread::spawn返回的JoinHandle，用于下面的join等待
    // 可在闭包前面添加 move 关键字，这样闭包就会获取所有被引用变量的所有权
    let handle = thread::spawn(move || {
        for i in 1..10 {
            println!("hi number {i} from the spawned thread!");
            thread::sleep(Duration::from_millis(1));
        }
    });

    // 主线程
    for i in 1..5 {
        println!("hi number {i} from the main thread!");
        thread::sleep(Duration::from_millis(1));
    }

    // join 会阻塞当前线程直到 handle 所代表的线程结束
    // 不join则主线程结束时子线程会被强制结束
    handle.join().unwrap();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// use std::thread; // 前面已经引入了
use std::sync::{Arc, Barrier};
fn test_barriers() {
    // 数组用于保存线程句柄
    let mut handles = Vec::with_capacity(6);
    // 创建一个线程屏障，等待6个线程，通过线程安全的Arc智能指针来共享
    let barrier = Arc::new(Barrier::new(6));

    for _ in 0..6 {
        let b = barrier.clone();
        handles.push(thread::spawn(move|| {
            println!("before wait");
            // 增加一个线程屏障
            b.wait();
            println!("after wait");
        }));
    }

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
use std::sync::{Mutex, Condvar};
fn test_condvar() {
    // 创建一个Mutex和Condvar的元组，通过Arc原子引用计数智能指针来共享
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    // 通过clone来克隆一个智能指针，引用计数+1
    let pair2 = pair.clone();

    thread::spawn(move|| {
        let (lock, cvar) = &*pair2;
        // 获取锁，并且获取锁保护的值
            // lock.lock()返回的是一个Result<MutexGuard<bool>, Error>
            // 由于MutexGuard是一个结构体，它实现了Deref和DerefMut trait，可以像引用一样使用
            // 因此可以直接将MutexGuard<bool>赋值给mut started，而不需要显式地解引用
        // 暂时使用unwrap()处理错误，生产代码中不推荐，会导致程序在遇到错误时panic
        let mut started = lock.lock().unwrap();
        println!("changing started");
        // 修改共享状态：started的值设置为true
        *started = true;
        // 通知等待的线程，此处的cvar和主线程中的cvar是同一个
        cvar.notify_one();
    });

    let (lock, cvar) = &*pair;
    // 获取锁
    let mut started = lock.lock().unwrap();
    // 等待started的值变为true，如果为false则通过条件变量的wait函数挂起当前线程
    while !*started {
        // wait函数等待满足条件
            // wait返回一个Result<MutexGuard<T>, Error>，该方法会释放Mutex的锁，并使当前线程进入等待状态
            // 当其他线程调用notify_one或notify_all方法通知条件变量时，wait方法会重新获取锁，并返回一个新的MutexGuard
        started = cvar.wait(started).unwrap();
    }

    println!("started changed");
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// use std::thread;
use std::sync::Once;
static mut VAL: usize = 0;
static INIT: Once = Once::new();
fn test_call_once() {
    let handle1 = thread::spawn(move || {
        INIT.call_once(|| {
            unsafe {
                VAL = 1;
            }
        });
    });

    let handle2 = thread::spawn(move || {
        INIT.call_once(|| {
            unsafe {
                VAL = 2;
            }
        });
    });

    handle1.join().unwrap();
    handle2.join().unwrap();

    println!("{}", unsafe { VAL });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
fn test_mpsc() {
    // 创建一个消息通道，返回一个元组 (发送者, 接收者)
    let (tx, rx) = mpsc::channel();
    // 也可显式指定泛型类型
    let (tx2, rx2): (mpsc::Sender<String>, mpsc::Receiver<String>) = mpsc::channel();

    thread::spawn(move || {
        // 发送一个数字1, send方法返回Result<T,E>，通过unwrap进行快速错误处理
        tx.send(1).unwrap();

        // 传输未实现Copy trait的类型
        let s = String::from("hello");
        tx2.send(s).unwrap();
        // 无法再使用s，因为所有权已经转移，报错：value borrowed here after move
        // println!("s: {}", s);
    });

    // 在主线程中接收子线程发送的消息并输出
    // 若接收不到消息，recv方法会阻塞当前线程，直到读取到值，或者通道被关闭
    let recv = rx.recv().unwrap();
    // 使用 try_recv 方法则不会阻塞当前线程，若接收不到消息时，返回一个错误
    // let recv = rx.try_recv().unwrap();
    println!("recv: {}", recv);

    let s = rx2.recv().unwrap();
    println!("s: {}", s);
}

1
2
3
4
5
6
let (tx, rx) = mpsc::channel();
...
// 循环阻塞的从rx迭代器中接收消息
for received in rx {
    println!("Got: {}", received);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
fn test_mutex() {
    // 使用`Mutex`结构体的关联函数创建新的互斥锁实例
    let m = Mutex::new(5);

    {
        // 获取锁，然后deref为`m`的引用
        // lock返回的是Result
        let mut num = m.lock().unwrap();
        *num = 6;
        // 锁自动被drop
    }

    println!("m = {:?}", m);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// use std::rc::Rc;
use std::sync::Arc;
use std::thread;
fn test_threads_mutex() {
    // 创建一个整数0，并包装在Mutex中，然后包装在Arc指针中进行共享，Mutex保证并发安全
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        // clone增加引用计数并克隆 Arc 的所有权，这样每个线程都有自己的 Arc 实例指向相同的 Mutex
        let counter = Arc::clone(&counter);
        // 创建线程
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            // Deref自动解引用，获得指向Mutex内部数据的引用
            *num += 1;
        });
        handles.push(handle);
    }

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

    // 结果为 Result: 10，10个线程都+1
    println!("Result: {}", *counter.lock().unwrap());
}

1
2
3
4
5
fn test_dead_lock() {
    let data = Mutex::new(0);
    let d1 = data.lock();
    let d2 = data.lock();
}

1
2
3
4
5
6
7
8
9
10
11
12
fn test_rwlock() {
    let lock = RwLock::new(5);
    {
        // 同时多个读锁
        let r1 = lock.read().unwrap();
        let r2 = lock.read().unwrap();
    }
    {
        // 同时仅支持一个写锁
        let mut w = lock.write().unwrap();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
use std::sync::atomic::{AtomicU64, Ordering};

const N_TIMES: u64 = 10000000;
const N_THREADS: usize = 10;

static R: AtomicU64 = AtomicU64::new(0);
fn add_n_times(n: u64) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            R.fetch_add(1, Ordering::Relaxed);
        }
    })
}
fn test_atomic() {
    let mut threads = Vec::with_capacity(N_THREADS);

    for _ in 0..N_THREADS {
        threads.push(add_n_times(N_TIMES));
    }
    for thread in threads {
        thread.join().unwrap();
    }
    print("R:{}", R.load(Ordering::Relaxed));
}

1
2
3
4
5
6
7
8
9
// Rc源码片段
// Rc<T>的Send和Sync特征被特地移除了实现
impl<T: ?Sized> !marker::Send for Rc<T> {}
impl<T: ?Sized> !marker::Sync for Rc<T> {}

// Arc源码片段
// 实现了Sync + Send
unsafe impl<T: ?Sized + Sync + Send> Send for Arc<T> {}
unsafe impl<T: ?Sized + Sync + Send> Sync for Arc<T> {}