Rust 并发编程基础

线程基础

每个程序都以一个主线程开始启动。主线程可以生成一个新线程，该线程将成为其子线程，子线程可以进一步产生自己的线程。

Rust 中最简单的多线程大概就是使用 thread::spawn:

use std::thread;

fn fibonacci(n: u32) -> u32 {
    if n == 0 {
        return 0;
    } else if n == 1 {
        return 1;
    } else {
        return fibonacci(n - 1) + fibonacci(n - 2);
    }
}

fn main() {
    thread::spawn(|| {
        let f = fibonacci(30);
        println!("Hello from a thread!. fibonacci(30) = {}", f);
    });
    println!("Hello, world!");
}

运行这段代码，会发现闭包内的函数结果并没有被输出：

Hello, world!

子线程是以分离状态创建的，程序执行到 println! 后就结束了，而相对耗时的子线程并没有结束。如果想要看到正确的结果，需要等待子线程。thread::spawn 返回一个 JoinHandle 类型的值，可以将它存放到变量中。这个类型相当于子线程的句柄，用于连接线程。如果忽略它，就没有办法等待线程。

fn main() {
    let child = thread::spawn(|| {
        let f = fibonacci(30);
        println!("Hello from a thread!. fibonacci(30) = {}", f);
        f
    });

    println!("Hello, world!");
    let v = child.join().expect("Could not join child thread");
    println!("value: {:?}", v);
}

join 返回一个 Result，可以使用 expect 方法来获取返回值。这样主线程就会等待子线程完成，而不会先结束程序了，就可以看到我们想要的结果：

Hello, world!
Hello from a thread!. fibonacci(30) = 832040
value: 832040

自定义线程

还可以通过 thread 中的 Builder 设置线程的属性来配置线程的 API，例如名称或者堆栈的大小。如果有panic! 会输出对应的调试信息，就可以看到线程名称。

fn main() {
    let my_thread = Builder::new()
    .name("my_thread".to_string())
    .stack_size(1024 * 4)
    .spawn(move || {
        let v = fibonacci(30);
        println!("fibonacci(30) = {}", fibonacci(30));
        v
    });

    println!("Hello, world!");
    let v = my_thread.unwrap().join().expect("Could not join child thread");
    println!("value: {:?}", v);
}

访问线程中的数据

不和主线程交互的子程序是非常少见的。一种常见的应用模式是使用多线程访问列表中的元素来执行某些运算。

考虑以下代码：

use std::thread;

fn main() {
    let v = vec![1, 3, 5, 7, 9];
    for n in v {
        thread::spawn(move || {
            println!("{}", n * n);
        });
    };
}

这里的 move 是必要的，否则你没法保证主线程中 v 会不会在线程之前被销毁。使用 move 之后，所有权转移到了子线程内，从而使得不会出现因为生命周期造成的数据无效的情况。

上述代码虽然可以正常运行，但是子线程可能的执行的周期可能比主线程还长。因此很可能出现结果还没有完全打印出来，就已经结束的情况。

在控制台随机运行三次，出现的结果：

1     1     1
9     9     25
49    81    9
      49    49

要解决这个问题，最简单的方式是在存下线程句柄，并在最后使用 join 阻塞主线程。

use std::thread;

fn main() {
    let v = vec![1, 3, 5, 7, 9];
    let mut childs = vec![];
    for n in v {
        let c = thread::spawn(move || {
            println!("{}", n * n);
        });
        childs.push(c);
    };

    for c in childs {
        c.join().unwrap();
    }
}

线程的并发模型

我们使用线程的主要目的是执行可以拆分为多个子问题的任务，其中线程可能需要彼此通信或共享数据。并发模型指定多线程之间如何进行指令交互和数据共享，以及它们在时间和空间（这里指内存）上如何完成进度。

Rust 内置了两种流行的并发模型：通过同步共享数据和通过消息传递共享数据。

状态共享模型

上面的代码中，使用了 move，意味着在之后的代码里不能再访问该数据，而且无法和其它线程共享所有权。单线程中，可以使用Rc 共享所有权，多线程则使用的是 Arc：

use std::thread;
use std::sync::Arc;

fn main() {
    let v = Arc::new(vec![1, 3, 5, 7, 9]);
    let mut childs = vec![];
    for n in 0..v.len() {
        let ns = v.clone();
        let c = thread::spawn(move || {
            println!("{}", ns[n] * ns[n]);
        });
        childs.push(c);
    };

    for c in childs {
        c.join().unwrap();
    }
}

状态共享之后，是不能直接修改共享的数据的，如果尝试写入值，就会发现报错，例如：

// ...
    for n in 0..v.len() {
        let ns = v.clone();
        let c = thread::spawn(move || {
            println!("{}", ns[n] * ns[n]);
            // 试图修改
            v.push(ns[n] * ns[n]);
        });
        childs.push(c);
    };
// ...

复制 Arc 分发了对内部值的不可变引用。要改变来自多线程的数据，我们需要使用一种提供共享可变性的类型，就像 RefCell 那样。但与 Rc 类似，RefCell 不支持多线程。

因此，我们需要使用它们的线程安全的变体，例如 Mutex(互斥) 或 RwLock（读写锁） 包装器类型。

互斥

互斥锁（mutex）是 mutual 和 exclusion 的缩写，是一种广泛使用的同步原语，用于确保一段代码一次只能有一个线程执行。

可以使用 Mutex 配合 Arc 实现多线程共享数据的修改。

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

fn main() {
    let v = Arc::new(Mutex::new(vec![1, 3, 5, 7, 9]));
    let mut childs = vec![];
    let mut length = v.lock().unwrap().len();

    for n in 0..length {
        let ns = v.clone();
        let c = thread::spawn(move || {
            let mut vs = ns.lock().unwrap();
            let val = vs[n];
            vs.push(val * val);
        });
        childs.push(c);
    };

    for c in childs {
        c.join().unwrap();
    }

    length = v.lock().unwrap().len();

    for n in 0..length {
        println!("{}", v.lock().unwrap()[n]);
    }
}

在互斥锁上执行锁定将阻止其他线程调用锁定，直到锁定消失为止。因此，以一种细粒度的方式构造代码是很重要的。

还有一种与互斥锁类似的替代方法，即 RwLock 类型，它对类型的锁定更敏感，并且在读取比写入更频繁的情况下性能更好。

RwLock

互斥锁适用于大多数应用场景，但对于某些多线程环境，读取的发生频率高于写入的。在这种情况下，我们可以采用 RwLock 类型。

RwLock 表示 Reader-Writer 锁。通过 RwLock，我们可以同时支持多个读取者，但在给定作用域内只允许一个写入者。这比互斥锁要好得多，互斥锁对线程所需的访问类型是未知的。

use std::thread;
use std::sync::{ Arc, RwLock};

fn main() {
    let v = Arc::new(RwLock::new(vec![1, 3, 5, 7, 9]));
    let mut childs = vec![];
    let mut length = v.read().unwrap().len();

    for n in 0..length {
        let vs = v.clone();
        let c = thread::spawn(move || {
            let val = vs.read().unwrap()[n];
            let mut ns = vs.write().unwrap();
            ns.push(val * val);
        });
        childs.push(c);
    };

    for c in childs {
        c.join().unwrap();
    }

    length = v.read().unwrap().len();

    for n in 0..length {
        println!("{}", v.read().unwrap()[n]);
    }
}

但是在某些操作系统（如: Linux）上 RwLock 会遇到写入者 饥饿问题 。这种情况是因为读取者不断访问共享资源，从而导致写入者线程永远没有机会访问共享资源。

通过消息传递进行通信

线程还可以通过被称为消息传递的更高级抽象来互相通信。这种线程通信模型 避免了需要用户显式锁定的要求。

标准库中的 std::sync::mpsc 模块提供了一个无锁的 多生产者、单订阅者（消费者）队列 ，以此作为希望彼此通信的线程的共享消息队列。mpsc 模块标准库包含两种通道。

• channel：这是一个异步的无限缓冲通道。
• sync_channel：这是一个同步的有界缓冲通道。

主线程生成值 0,1,…,9，然后在新生成的线程输出它们：

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

fn main() {
    let (tx, rx) = channel();
    let join_handle = thread::spawn(move || {
        while let Ok(msg) = rx.recv() {
            println!("Received {}", msg);
        }
    });

    for i in 0..10 {
        tx.send(i).unwrap();
    }

    join_handle.join().unwrap();
}

首先调用了 channel 方法，这将返回两个值 tx 和 rx。tx 是包含 Sender<T> 类型的发送端，rx 是包含 Receiver<T> 类型的接收端。它们的名字是约定俗成的。

多个生产者，单个消费者（Multi Producer, Single Consumer，MPSC）方法提供了多个作者（send），但只提供了一个读者（join_handle）。

使用默认的异步通道时，send 方法永远不会阻塞。这是因为通道缓冲区是无限的，所以总是会提供更多的空间。当然，它实际上并不是无限的，只是在概念上如此：如果你在没有收到任何数据的情况下向通道发送数千兆字节，那么系统可能会耗尽内存。

同步通道有一个有界缓冲区，当它被填满时，send方法会被阻塞，直到通道中出现更多空间。其他用法和异步通道类似：

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

fn main() {
    let (tx, rx) = mpsc::sync_channel(1);
    let tx_clone = tx.clone();

    let _ = tx.send(0);

    thread::spawn(move || {
        let _ = tx.send(1);
    });

    thread::spawn(move || {
        let _ = tx_clone.send(2);
    });

    println!("{:?}", rx.recv().unwrap());
    println!("{:?}", rx.recv().unwrap());
    println!("{:?}", rx.recv().unwrap());
    println!("{:?}", rx.recv());
}

同步通道大小为 1 时，这意味着通道中不可能存在多个元素。在这种情况下，第一次发送请求之后的任何请求都会被阻塞。如果通道是空的，那么 recv 调用会返回 Err 值。

代码的执行结果：

0
1
2
Err(RecvError)

小结

单线程环境中防止内存安全违规的相同所有权规则也适用于具有标记特征组合的多线程环境。