Rust 多執行緒與平行處理：通道、Mutex 與 RwLock 應使用案例項

– 8 分鐘閱讀

Rust 提供了多種多執行緒和平行處理機制，讓開發者能夠有效利用多核心處理器的效能。std::thread 和 scoped threads 提供了建立新執行緒的方式，而通道則允許執行緒之間進行資料交換。同步通道適用於確保資料傳輸的順序性，而非同步通道則更注重效率。Mutex 提供了互斥鎖，確保同一時間只有一個執行緒能夠存取分享資料，防止資料競爭。RwLock 則更進一步，允許多個執行緒同時讀取資料，但在寫入時會鎖定資源，兼顧效率和資料安全。這些機制與 Arc 智慧指標結合使用，可以更安全地管理分享資料，避免懸垂指標等問題。

多執行緒

Rust 提供了兩種多執行緒的方式：標準函式庫的 std::thread 和 scoped threads。

標準函式庫的 std::thread

標準函式庫的 std::thread 提供了一種基本的多執行緒方式。您可以使用 thread::spawn 函式建立一個新執行緒，並使用 thread::join 函式等待執行緒完成。

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        println!("Hello from new thread!");
    });
    handle.join().unwrap();
}

Scoped threads

Scoped threads 是 Rust 1.63.0 中引入的一種新特性。它允許您在一個作用域中建立多個執行緒，並自動等待所有執行緒完成。

use std::thread::scope;

fn main() {
    scope(|s| {
        s.spawn(|| {
            println!("Hello from new thread!");
        });
    });
}

通道

Rust 提供了兩種通道的方式：同步通道和非同步通道。

同步通道

同步通道使用 std::sync::mpsc::sync_channel 函式建立。它是一種同步的多生產者和單消費者佇列。

use std::sync::mpsc::sync_channel;

fn main() {
    let (tx, rx) = sync_channel(0);
    tx.send("Hello").unwrap();
    println!("Received: {}", rx.recv().unwrap());
}

非同步通道

非同步通道使用 std::sync::mpsc::channel 函式建立。它是一種非同步的多生產者和單消費者佇列。

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

fn main() {
    let (tx, rx) = channel();
    tx.send("Hello").unwrap();
    println!("Received: {}", rx.recv().unwrap());
}

通道的使用

以下是使用通道的範例：

use std::thread;
use std::sync::mpsc::channel;
use std::io::{Result, stdin};

fn main() -> Result<()> {
    let (tx, rx) = channel();
    let handle = thread::spawn(move || {
        match rx.recv() {
            Ok(msg) => println!("Received: {}", msg),
            Err(e) => eprintln!("{}", e.to_string()),
        }
    });
    println!("Please enter a message to send: ");
    let mut input = String::new();
    stdin().read_line(&mut input)?;
    tx.send(input.trim().to_owned()).unwrap();
    handle.join().unwrap();
    Ok(())
}

這個範例建立了一個通道，然後在一個新執行緒中接收訊息。主執行緒等待使用者輸入一個訊息，然後傳送訊息給新執行緒。新執行緒接收到訊息後，列印預出來。

使用 Mutex 實作多執行緒分享資料

在 Rust 中，Mutex（Mutual Exclusion）是一種鎖機制，允許多個執行緒安全地存取分享資料。以下是使用 Mutex 實作多執行緒分享資料的範例。

Mutex 的基本使用

Mutex 的基本使用方法是使用 lock() 方法鎖定資料，然後使用 unlock() 方法解鎖資料。然而，在 Rust 中，Mutex 的鎖定和解鎖是自動的，無需手動呼叫 unlock() 方法。

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter_clone.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Final counter value: {}", *counter.lock().unwrap());
}

使用 Mutex 儲存學生成績

以下是使用 Mutex 儲存學生成績的範例。

use std::io::{stdin, Result};
use std::sync::{Arc, Mutex};
use std::thread;

type Students = Arc<Mutex<Vec<u32>>>;

fn new_empty_students() -> Students {
    Arc::new(Mutex::new(Vec::new()))
}

fn get_grade(student_num: u32) -> u32 {
    println!("Please enter grade for student {}: ", student_num);
    let mut input = String::new();
    stdin().read_line(&mut input).unwrap();
    input.trim().parse().unwrap()
}

fn main() -> Result<()> {
    let students = new_empty_students();
    let mut join_handles = Vec::new();

    println!("Please enter number of students: ");
    let mut num_input = String::new();
    stdin().read_line(&mut num_input)?;
    let num_stu: u32 = num_input.trim().parse().unwrap();

    for i in 0..num_stu {
        let mut s = students.clone();
        let grade = get_grade(i + 1);
        let t = thread::spawn(move || {
            let mut s = s.lock().unwrap();
            s.push(grade)
        });
        join_handles.push(t);
    }

    let t = thread::spawn(move || {
        let s = students.lock().unwrap();
        for i in 0..s.len() {
            println!("Student {} => Grade: {}", i + 1, s[i])
        }
    });
    join_handles.push(t);

    for jh in join_handles {
        jh.join().unwrap();
    }

    Ok(())
}

結果

執行上述程式後，會要求使用者輸入學生人數和每個學生的成績，然後會印出每個學生的成績。

Please enter number of students: 
2
Please enter grade for student 1: 
98
Please enter grade for student 2: 
69
Student 1 => Grade: 98
Student 2 => Grade: 69

使用RwLock實作多執行緒讀寫鎖

RwLock是一種多執行緒鎖，允許多個執行緒同時讀取資料，而寫入資料時則需要獨佔鎖。這種鎖可以提高多執行緒程式的效率，特別是在讀取操作遠多於寫入操作的情況下。

RwLock的優點

  • 讀取操作可以並發，提高了程式的效率
  • 寫入操作需要獨佔鎖，確保資料的一致性

RwLock的使用

以下是RwLock的基本使用方法：

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

// 建立一個RwLock例項
let lock = Arc::new(RwLock::new(0));

// 讀取鎖
let read_lock = lock.read().unwrap();
println!("Read: {}", *read_lock);

// 寫入鎖
let mut write_lock = lock.write().unwrap();
*write_lock = 10;
println!("Write: {}", *write_lock);

RwLock的實作

以下是RwLock的實作示例：

use std::fs::{read_dir, FileType, read_to_string};
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use std::thread;

// 定義一個Info結構體
#[derive(Debug, Clone)]
struct Info {
    type_: FileType,
    content: String,
    path: PathBuf,
}

impl Info {
    // 建立一個新的Info例項
    pub fn new(type_: FileType, content: String, path: PathBuf) -> Self {
        Self { type_, content, path }
    }
}

// 定義一個Infos型別別名
type Infos = Arc<RwLock<Vec<Info>>>;

// 建立一個新的Infos例項
fn new_infos() -> Infos {
    Arc::new(RwLock::new(Vec::new()))
}

fn main() {
    // 讀取目錄
    let dir = read_dir("important").unwrap();
    // 建立一個新的Infos例項
    let infos = new_infos();

    // 建立一個新的執行緒
    let t = thread::spawn(move || {
        // 克隆infos例項
        let i = infos.clone();
        // 遍歷目錄
        for d in dir {
            // 解封裝目錄條目
            let d = d.unwrap();
            // 取得檔案型別
            let type_ = d.file_type().unwrap();
            // 初始化內容
            let mut content = String::new();
            // 如果是目錄，設定內容為"Directory"
            if type_.is_dir() {
                content.push_str("Directory");
            } else {
                // 如果是檔案，讀取檔案內容
                content = read_to_string(d.path()).unwrap();
            }
            // 取得檔案路徑
            let path = d.path();
            // 建立一個新的Info例項
            let info = Info::new(type_, content, path);
            // 取得寫入鎖
            let mut i = i.write().unwrap();
            // 推入Info例項到向量中
            i.push(info);
        }
        // 取得讀取鎖
        let updated = i.read().unwrap();
        // 遍歷更新的向量
        let u = updated.iter().map(|x| {
            // 取得檔案型別
            let mut type_ = String::new();
            if x.type_.is_dir() {
                type_.push_str("Directory");
            } else {
                type_.push_str("File");
            }
            // 取得檔案路徑
            let path = x.path.to_str().unwrap();
            println!("Path: {} | File type: {} \nContent: \n{}\n", path, type_, x.content);
        }).collect::<()>();
        // 傳回集合
        u
    });

    // 等待執行緒完成
    t.join().unwrap();
}

Rust 的並發安全機制

Rust 的並發安全機制是根據所有權和借用系統的。這個系統確保了在多個執行緒之間分享資料的安全性。Rust 的標準函式庫提供了多種工具和資料結構來支援並發程式設計，包括 Arc、Mutex、RwLock 等。

Arc 和分享資料

Arc（Atomic Reference Counting）是一種用於分享資料的智慧指標。它允許多個執行緒安全地分享同一份資料。當最後一個參照 Arc 的執行緒結束時，Arc 會自動釋放資料。

use std::sync::Arc;

fn main() {
    let shared_data = Arc::new(10);
    let thread = std::thread::spawn({
        let shared_data = Arc::clone(&shared_data);
        move || {
            println!("Thread: {}", *shared_data);
        }
    });
    println!("Main: {}", *shared_data);
    thread.join().unwrap();
}

Mutex 和互斥鎖

Mutex（Mutual Exclusion）是一種用於保護分享資源的互斥鎖。它允許只有一個執行緒在同一時間記憶體取分享資源。

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter_clone.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap());
}

RwLock 和讀寫鎖

RwLock（Read-Write Lock）是一種用於保護分享資源的讀寫鎖。它允許多個執行緒同時讀取分享資源，但只允許一個執行緒寫入分享資源。

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

fn main() {
    let data = Arc::new(RwLock::new(10));
    let mut handles = vec![];

    for _ in 0..10 {
        let data_clone = Arc::clone(&data);
        let handle = thread::spawn(move || {
            let num = data_clone.read().unwrap();
            println!("Thread: {}", *num);
        });
        handles.push(handle);
    }

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

    let mut num = data.write().unwrap();
    *num = 20;
    println!("Result: {}", *num);
}

練習題

  1. 建立一個留言板應用程式，允許使用者提交留言和檢視所有留言。
  2. 建立一個 RSS 讀取器，使用 rss 函式函式庫解析 RSS 訂閱，並使用 crossbeam 函式函式庫進行執行緒間通訊。

答案

  1. 留言板應用程式可以使用 rocket 函式函式庫建立 Web 伺服器，使用 serde 函式函式庫序列化和反序列化資料。
use rocket::response::Redirect;
use rocket::Request;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Message {
    name: String,
    message: String,
}

#[post("/message", data = "<message>")]
fn create_message(message: Message) -> Redirect {
    // 儲存留言到資料函式庫
    Redirect::to(uri!("/messages"))
}

#[get("/messages")]
fn get_messages() -> String {
    // 讀取所有留言從資料函式庫
    let messages = vec![];
    format!("{}", messages)
}
  1. RSS 讀取器可以使用 rss 函式函式庫解析 RSS 訂閱，並使用 crossbeam 函式函式庫進行執行緒間通訊。
use rss::Channel;
use crossbeam::channel;

fn main() {
    let (tx, rx) = channel::unbounded();
    let mut channel = Channel::new("https://example.com/rss");
    channel.update().unwrap();

    for item in channel.items() {
        let title = item.title().unwrap();
        let link = item.link().unwrap();
        tx.send((title, link)).unwrap();
    }

    let mut handles = vec![];
    for _ in 0..10 {
        let rx_clone = rx.clone();
        let handle = std::thread::spawn(move || {
            let (title, link) = rx_clone.recv().unwrap();
            println!("Title: {}, Link: {}", title, link);
        });
        handles.push(handle);
    }

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

Rust 網頁應用程式開發

導言

本文將介紹如何使用 Rust 開發一個簡單的網頁應用程式，該程式將允許使用者提交訊息，並將其顯示在網頁上。

訊息結構體

首先，我們需要定義一個結構體來代表使用者提交的訊息。這個結構體將包含使用者的名稱、訊息內容和接收時間。

#[derive(Debug, Clone, Serialize, Deserialize, FromForm)]
pub struct Message {
    pub name: String,
    pub message: String,
    pub received_at: Option<String>,
}

訊息相關函式

接下來，我們需要實作一些函式來處理訊息。其中包括更新訊息的接收時間、將訊息轉換為 HTML 程式碼等。

impl Message {
    pub fn received(&mut self) {
        self.received_at = Some(Utc::now().to_string());
    }

    pub fn to_html(&self) -> String {
        // ...
    }
}

訊息儲存和讀取

我們需要實作函式來儲存和讀取訊息。這些函式將使用 JSON 格式來儲存訊息。

pub fn get_messages() -> Vec<Message> {
    from_str(&read_to_string("messages.json").unwrap()).unwrap()
}

pub fn rewrite_messages(messages: Vec<Message>) -> Result<()> {
    // ...
}

網頁應用程式

現在，我們可以開始實作網頁應用程式了。首先，我們需要建立一個函式來處理使用者提交的表單。

#[post("/", data = "<message>")]
pub fn post_message(message: Form<Message>) -> Redirect {
    // ...
}

接下來，我們需要實作一個函式來顯示網頁內容。

#[get("/")]
pub async fn index() -> Option<NamedFile> {
    // ...
}

最後，我們需要實作主函式來啟動網頁應用程式。

pub fn rocket() {
    // ...
}

建立RSS Feed的Rust專案

本文將介紹如何使用Rust語言建立一個RSS feed的專案。首先，我們需要建立一個新的Rust專案，並新增必要的依賴項。

新增依賴項

在Cargo.toml檔案中新增以下依賴項：

[dependencies]
rocket = "0.5.0-rc.2"
rss = "2.0.1"
crossbeam = "0.8.2"

這些依賴項分別是：

  • rocket: 一個Rust的Web框架
  • rss: 一個Rust的RSS解析函式庫
  • crossbeam: 一個Rust的並發函式庫

建立Feed結構

接下來，我們需要建立一個Feed結構來代表每個RSS feed的文章：

#[derive(Debug, Clone)]
pub struct Feed {
    pub title: String,
    pub pub_date: String,
    pub link: String,
    pub categories: Vec<String>,
    pub description: String,
}

這個結構有五個欄位：title、pub_date、link、categories和description。

從技術架構的視角來看，Rust 的多執行緒模型和通道機制為開發者提供了構建高效能併發應用程式的強大工具。本文深入探討了std::thread、scoped threads、同步和非同步通道的特性及使用方法，並佐以 Mutex 和 RwLock 等機制，展示了 Rust 如何有效管理多執行緒分享資料，避免資料競爭和死鎖等問題。然而，Rust 的所有權和借用系統雖然保障了記憶體安全，但也增加了程式碼的複雜度，開發者需要仔細設計程式碼結構，才能充分發揮 Rust 的併發效能優勢。展望未來，隨著非同步程式設計模型的日益普及，Tokio 等非同步執行時與 Rust 的整合將進一步提升其在高併發場景下的應用潛力，同時也需要更多針對性的效能最佳化工具和最佳實務的出現。對於追求極致效能且重視程式碼安全的開發者而言，Rust 的多執行緒程式設計模型無疑是一個值得深入研究和應用的方向。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。