Rust中文翻译20

4.6 并发

Page 100

并发和并行在计算机科学中是非常重要的主题.在工业领域也很火.计算机如今有越来越多的核心,然而很多程序员还没有准备好使用它们.

Rust安全的内存特性同样适用于并发存储.并发的Rust代码也是内存安全的,没有数据竞争.Rust的类型系统会保证这一点,给你提供了有利的帮助在编译时实现并发代码.

在我们开始讨论Rust的并发之前,我们需要理解一个很重要的事情:Rust是一个低级别语言,它的所有功能都是通过库来提供的,而不是语言自身.也就是说你不喜欢Rust处理并发的方式,你完全可以实现一个你自己的版本.mio(https://github.com/carllerche/mio)就是一个真实世界的例子.

Page 101

背景:Send和Sync

并发是一个很难使用的技术.在Rust中,我们有一个很强的静态类型系统来帮助我们使用并发代码.为此,Rust提供了两种模板来实现并发.

Send: 第一种就是我们将要讨论的Send.当一个类型T实现了Send,它就告诉了编译器这个类型可以在多个线程安全传递变量所有性.

设置一些限制是很重要的.比如,如果我们有一个链接两个线程的通道,我们希望可以沿着通道将数据从一个线程传递到另一个线程.因此,我们要确保Send可以被这个类型实现.

相反,如果我们用FFI包装了一个不是线程安全的库,我们就不会要实现Send,编译器也就会帮我们确保它不会离开当前线程.

Sync: 第二种就是Sync.当一个类型T实现了Sync,它就告诉了编译器这个类型在使用多线程的时候不会引入不安全的内存.

例如,共享一个不可变数据的原子引用计数就是类型安全的.Rust提供了一个类型Arc<T>,他就实现了Sync,所以它可以安全的在两个线程之间共享.

这两种模板可以保证你的并发代码工作正确.在我们解释原因之前,我们需要先学习一下如何创建一个Rust并发程序!

线程:

Rust标准库提供了一个线程库,它可以运行并发的代码.这有一个基本的例子来使用std::thread:

use std::thread;

fn main() {
thread::spawn(|| {
println!("Hello from a thread!");
});
}

thread::spawn()函数接受一个闭包,这个闭包在一个新的线程被执行.它返回一个该线程的句柄,它可以等到子线程结束的时候来执行它的结果:

use std::thread;

fn main() {
let handle = thread::spawn(|| {
"Hello from a thread!"
});

println!("{}", handle.join().unwrap());
}

许多语言都有执行多线程的能力,但是大部分都不是安全的.有很多书都讲了如果防止共享可变数据的时候发生错误.Rust类型系统也提供了这个能力,它是从编译时就保证了不会发生数据竞争.让我们来讨论一下如果再多线程之间共享数据吧.

安全共享可变状态:

基于Rust的类型系统,我们有一个听起来像谎言的概念:"安全共享可变状态."许多程序员都同一个观点:共享可变状态是非常非常糟糕的.

有人曾经这样说:

共享可变状态是万恶之源.大多数语言都尝试用"可变的"部分来处理它,但是Rust语言用"共享的"来处理它.

Rust语言的所有权系统可以防止指针的误用,同样可以防止数据竞争,那是并发中最糟糕的bug.

作为一个例子,这有一个Rust程序,在多数语言中会发生数据竞争.它无法编译:

use std::thread;

fn main() {

let mut data = vec![1u32, 2, 3];

for i in 0..3 {

thread::spawn(move || {

data[i] += 1;

});

}

thread::sleep_ms(50);

}

这会有一个错误:

Page 103

在此例中,我们知道我们的代码是安全的,但是Rust不能确保.而且它实际上并不安全:如果我们在每个线程都有一个data的引用,该线程又取得了引用的所有权,那么我们就有3个所有者了!这很糟.我们可以通过Arc<T>类型来解决这个问题,它是一个原子的引用计数指针.原子的意思是它在多线程共享的时候是安全的.

Arc<T>提供了一个额外的属性来确保它在多线程共享的时候是安全的:它认为它的内容是Sync.但是在我们的例子中,我们需要改变变量的值.我们需要一个类型能保证只有一个人在同一时刻可以改变它的值.为此,我们可以试用Mutex<T>类型.第二个版本的例子仍然不能工作但是原因不一样:

use std::thread;

use std::sync::Mutex;

fn main() {

let mut data = Mutex::new(vec![1u32, 2, 3]);

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

thread::sleep_ms(50);

}

这里是错误信息:

Mutex和lock方法有如下签名:

fn lock(&self) -> LockResult<MutexGuard<T>>

因为Send没有被MutexGuard<T>实现,所以我们不能传递这个guard超越线程边界,所以我们得到了这个错误.

我们可以使用使用Arc<T>来解决.这个版本可以工作了:

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

fn main() {
let data = Arc::new(Mutex::new(vec![1u32, 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_ms(50);
}

我们现在在Arc类型上调用clone()方法,他可以增加内部计数.这个句柄会被移入新的线程.我们来看看线程内部的代码:

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

首先,我们调用lock()方法,它获得了mutex的锁.因为这个操作可能会失败,所以它返回了一个Result<T, E>类型,又因为这仅仅是个例子,我们直接调用unwrap()来得到data的一个引用.真实的代码应该包含更多的健壮的错误处理逻辑.因为有了锁,我们现在可以随意的改变它的值.

最后,当线程还在运行的时候,我们等待一段很短的时间.但是这还不理想:我们本可以选择一个更合理的时间等待,但是我们更有可能等的不够久或太久了,这取决于这个线程到底要多久才能完成我们的任务.

一个更精简的例子使用了Rust标准库提供的一个机制来同步多个线程.让我们来讨论其中的一种机制:channels.

Page 105

Channels:

这个例子使用了channels来同步多个线程,而不是等待一个特定的时间:

use std::sync::{Arc, Mutex};
use std::thread;
use std::sync::mpsc;
fn main() {
let data = Arc::new(Mutex::new(0u32));
let (tx, rx) = mpsc::channel();
for _ in 0..10 {
let (data, tx) = (data.clone(), tx.clone());
thread::spawn(move || {
let mut data = data.lock().unwrap();
*data += 1;
tx.send(());
});
}

for _ in 0..10 {
rx.recv();
}
}

我们使用mpsc::channel()方法来构造了一个channel.我们仅仅send了一个空的()到channel中,然后等待他们10个线程返回.

当这个channel正在发送通用信号的时候,我们就可以发送任何数据到channel中!

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

fn main() {
let (tx, rx) = mpsc::channel();
for _ in 0..10 {
let tx = tx.clone();
thread::spawn(move || {

let answer = 42u32;
tx.send(answer);
});
}
rx.recv().ok().expect("Could not receive answer");
}

一个u32被发送了.我们创建一个线程,他会计算结果,然后它通过channel会send()给我们答案.

Panics

一个Panic!会让当前执行线程崩溃.你可以试用Rust线程实现一个简单的孤立机制:

use std::thread;

let result = thread::spawn(move || {

panic!("oops!");

}).join();

assert!(result.is_err());