【第十一课】Rust并发编程（二）

前言

Channel

多生产者

前言

在上一节中，我们介绍了Rust中并发编程的方式之一：Fork和Join，通过新建线程提升代码的效率，这节课我们介绍并发编程的第二种方式：通道。Channel就类似于水管，通过Channel可以连接多个线程，达到多个线程之间协调作业。

Channel

我们以一个简单的需求为例来解释下Channel的使用方法。完成WordCount，使用2个线程，线程1从文件中读取数据，将数据通过Channel发送给线程2，线程2负责计算wordcount。

注意：

（1）下面代码中文件路径使用的是绝对路径

（2）使用mpsc::channel()可以创建一个通道，下面代码还指定了泛型，即在下面的代码中通道中只可以发送String类型的数据

（3）通道的返回值是一个发送者，一个消费者，发送者给线程1中的闭包，因为线程1负责读数据发送到通道中，消费者给线程2中的闭包，线程2负责读取数据处理wordcount。

（4）mpsc：指的是多生产者，单消费者的意思，即multi-producer，single-consumer。

rust">use std::collections::HashMap;
use std::sync::mpsc;
use std::thread::JoinHandle;
use std::{fs, io, thread};fn main() {let paths = vec!["/Users/xxxxx/rustproject/lesson11/src/test.txt".to_string()];let (receiver, handler1) = start_file_reader_thread(paths);let handler2 = start_file_word_count_thread(receiver);let _ = handler1.join().unwrap();let _ = handler2.join().unwrap();
}// 读取文件内容
fn start_file_reader_thread(documents: Vec<String>,
) -> (mpsc::Receiver<String>, JoinHandle<Result<(), io::Error>>) {let (sender, receiver) = mpsc::channel::<String>();let handle = thread::spawn(move || -> Result<(), io::Error> {for filename in documents {let text = fs::read_to_string(filename)?;if sender.send(text).is_err() {break;}}Ok(())});(receiver, handle)
}// word count
fn start_file_word_count_thread(texts: mpsc::Receiver<String>) -> JoinHandle<()> {let handle = thread::spawn(move || {// 处理let mut wc: HashMap<String, u32> = HashMap::new();for line in texts {let words: Vec<String> = line.split(" ").map(|x| x.to_string()).collect();for word in words {match wc.get(&word) {None => {wc.insert(word, 1);}Some(old) => {wc.insert(word, old + 1);}}}}// 打印for (word, cnt) in wc {println!("key = {}, count = {}", word, cnt);}});handle
}

多生产者

上面是单生产者的例子，我们扩展一下，使用多生产者。

注意：
（1）使用生产者的clone方法，扩展出多个生产者传递给不同的线程发送消息。

rust">use std::collections::HashMap;
use std::sync::mpsc;
use std::{fs, io, thread};fn main() {let path1 = vec!["/Users/xxx/rustproject/lesson11/src/test1.txt".to_string()];let path2 = vec!["/Users/xxx/rustproject/lesson11/src/test2.txt".to_string()];let (producer, consumer) = mpsc::channel::<String>();let producer1 = producer.clone();// 生产者let handler1 = thread::spawn(move || -> Result<(), io::Error> {for filename in path1 {let text = fs::read_to_string(filename)?;if producer.send(text).is_err() {break;}}Ok(())});let handler2 = thread::spawn(move || -> Result<(), io::Error>{for filename in path2 {let text = fs::read_to_string(filename)?;if producer1.send(text).is_err() {break;}}Ok(())});// 消费者let handler3 = thread::spawn(move || {// 处理let mut wc: HashMap<String, u32> = HashMap::new();for line in consumer {let words: Vec<String> = line.split(" ").map(|x| x.to_string()).collect();for word in words {match wc.get(&word) {None => {wc.insert(word, 1);}Some(old) => {wc.insert(word, old + 1);}}}}// 打印for (word, cnt) in wc {println!("key = {}, count = {}", word, cnt);}});//handler1.join();handler2.join();handler3.join();
}