无畏并发

并发

• Concurrent: 程序的不同部分之间独立的执行

• Parallel: 程序的不同部分之间同时执行

• Rust 无畏并发: 允许你编写没有细微Bug的代码, 并在不引入新bug的情况下易于重构

• 注意: 本章的内容泛指concurrent和parallel

使用线程来同时运行代码

进程与线程

• 在大部分OS里, 代码运行在进程(process)中, OS 同时管理多个进程

• 在你的程序里, 各部分可以分别同时运行, 运行这些独立部分的就是线程(thread)

• 多线程运行:

  • 提升性能表现
  • 增加复杂性, 无法保障各线程的执行顺序

多线程可导致的问题

• 竞争状态, 线程以不一致的顺序访问数据或资源
• 死锁, 两个线程彼此等待对方使用完其所拥有的资源, 线程无法继续
• 只在某些情况下发生的bug, 很难可靠地复制现象和修复

实现线程的方式

• 通过调用OS的API来创建线程1:1模型

  • 需要比较小的运行时

• 语言自己实现的线程(绿色线程)M:N模型

  • 需要更大的运行时

• Rust: 需要权衡运行时的支持

• Rust 标准库仅提供1:1模型的线程

通过spawn创建新线程

• 使用thread::spawn函数来创建新线程
  • 参数: 一个闭包(在新线程中运行的代码)

use std::thread;
use std::time::Duration;


fn main() {
    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));
    }
}

通过join Handle来等待所有线程完成

• thread::spawn 函数的返回值是 JoinHandle
• JoinHandle 持有值的所有权
  • 调用其join方法, 可以等待对应的其它线程的完成
• join方法: 调用handle 的join 方法会阻止当前运行的线程的执行,直到handle 所表示的这些线程的终结

使用move闭包

• move 闭包通常和thread::spawn一起使用, 它允许你使用其它线程的数据
• 在创建线程的时候, 把值的所有权从一个线程转移到另一个线程

use std::thread;
fn main() {
    let v = vec![1,2,3];
    let handle = thread::spawn(move ||  {
        println!("Here's a vector {:?}", v);
    });

    handle.join().unwrap();

}

使用消息传递来在线程间转移数据

消息传递

• 一种很流行且能保证安全并发的技术就说: 消息传递

  • 线程(或Actor)通过彼此发送消息(数据)来进行通信

• Go 语言的名言: 不要用共享内存来通信, 用通信来共享内存

• Rust: Channel (标准库提供)

Channel

• Channel 包含: 发送端,接收端
• 调用发送端的方法, 发送数据
• 接收端会检查和接收到达的数据
• 如果发送端, 接受端中任意一端被丢弃了, 那么Channel就关闭了

创建新的Channel

• 使用mpsc::channel函数创建一个新的channel
  • mpsc表示multiple producer, single consumer(多个生产者, 一个消费者)
  • 返回一个tuple(元组): 里面分别是发送端和接收端

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

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

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });
    let received = rx.recv().unwrap();

    println!("Got:{}", received);
    
}

接收端的方法

• • 一旦有值收到, 就返回Result<T, E>
  • 当发送端关闭时, 返回一个错误值

• try_recv 方法: 不会阻止线程, 直接返回一个Result<T, E>

  • Ok 值包含可用的信息
  • 否则, 返回一个错误值

• 通常会使用循环调用来检查try_recv的结果

Channel 和所有权转移

• 所有权在消息传递中非常重要: 能帮你编写安全, 并发的代码

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

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

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
        println!("val is {}", val);
    });
    let received = rx.recv().unwrap();

    println!("Got:{}", received);
    
}

println!("val is {}", val); 会报错: value borrowed here after move

发送多个值, 看到接收者在等待

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

fn main(){
    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_millis(200));
        }
    });

    for received in rx{
        println!("Got: {}", received);
    }   
}

通过克隆创建多个发送者

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

fn main(){
    let (tx, rx) = mpsc::channel();
    let tx1 = mpsc::Sender::clone(&tx);

    thread::spawn(move || {
        let vals = vec![
            String::from("1:hi"),
            String::from("1:from"),
            String::from("1:the"),
            String::from("1:thread")
        ];
        for val in vals{
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_millis(200));
        }
    });

    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_millis(200));
        }
    });

    for received in rx{
        println!("Got: {}", received);
    }
   
}

共享状态的并发

• Go 语言的名言: 不要用共享内存来通信, 用通信来共享内存

• Rust 支持通过共享状态来实现并发

• Channel 类似单所有权: 一旦将值的所有权转移至Channel, 就无法使用它了

• 共享内存并发类似多所有权, 多个线程可以同时访问同一块内存

使用Mutex来每次只允许一格线程来访问数据

• Mutex是mutual exclusion(互斥锁)的简写

• 在同一时刻, Mutex只允许一个线程来访问某些数据

• 想要访问数据

  • 线程必须首先获取互斥锁
    • lock 数据结构是mutex的一部分, 它能跟踪谁对数据拥有独占访问权
  • mutex 通常被描述为: 通过锁定系统来保护它所持有的数据

Mutex的两条规则

• 在使用数据之前, 必须尝试获得锁(lock)
• 使用完mutex所保护的数据, 必须对数据进行解锁, 以便其它线程可以获取锁

Mutex<T>的API

• 通过Mutex::new(数据)来创建Mutex<T>

  • Mutex<T> 是一个智能指针

• 访问数据前,通过lock方法来获取锁

  • 会阻塞当前线程
  • lock可能会失败
  • 返回的是 MutexGuard (智能指针, 实现了Deref和Drop)

use std::sync::Mutex;

fn main(){
    let m = Mutex::new(5);

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

多线程共享Mutex<T>

使用Arc<T>来在多个线程间共享所有权

• Arc<T>是atomically reference counted(原子引用计数)的缩写

• Arc<T> 和 Rc<T> 类似, 它可以用于并发场景

• 为什么所有的基础类型都不是原子的, 为什么标准库不默认使用Arc<T>?

  • 需要牺牲性能作为代价

• Arc<T> 和 Rc<T> 的API是相同的

use std::sync::{Mutex, Arc};
use std::thread;
fn main(){
    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());
    
}

RefCell<T>/Rc<T> vs Mutex<T>/Arc<T>

• Mutex<T> 提供了内部可变性, 和Cell家族一样
• 我们使用RefCell<T> 来改变Rc<T>里面的内容
• 我们使用Mutex<T> 来改变Arc<T>里面的内容
• 注意: Mutex<T>有死锁风险

使用Sync和Send trait的可扩展并发

Send 和 Sync trait

• Rust 语言的并发特性较少, 目前讲的并发新特性都来自标准库(而不是语言本身)
• 无需局限于标准库的并发, 可以自己实现并发
• 但在Rust语言中有两个并发的概念:
  • std::marker::Sync 和 std::marker::Send 这两个trait

Send：允许线程间转移所有权

• 实现Send trait 的类型可在线程间转移所有权
• Rust中几乎所有的类型都实现了Send
  • 但Rc<T>没有实现Send, 它只用于单线程场景
• 任何完全由Send类型组成的类型也被标记为Send
• 除了原始指针之外, 几乎所有的基础类型都是Send

Sync: 允许从多线程访问

• 实现Sync的线程可以安全地被多个线程引用
• 也就是说: 如果T是Sync, 那么 &T 就是Send
  • 引用可以被安全的送往另一个线程
• 基础类型都是Sync
• 完全由Sync类型组成的类型也是Sync
  • 但, Rc<T> 不是Sync的
  • RefCell<T> 和 Cell<T> 家族也不是Sync的
  • 而, Mutex<T> 是Sync的

手动来实现Send和Sync是不安全的

• 记住上面的话即可