快捷導(dǎo)航

Rust 中的 Tokio 線程同步機(jī)制詳解

更新時(shí)間：2025年09月23日 09:03:48 作者：天翼云開發(fā)者社區(qū)

Rust中的Tokio提供了多種線程同步機(jī)制,包括互斥鎖、讀寫鎖、屏障、信號(hào)量、通知及消息通道,用于協(xié)調(diào)并發(fā)訪問(wèn)、資源共享和狀態(tài)同步,提升程序的安全性和效率,本文給大家介紹Rust中的Tokio線程同步機(jī)制,感興趣的朋友一起看看吧

Rust 中的 Tokio 線程同步機(jī)制

在并發(fā)編程中，線程同步是一個(gè)重要的概念，用于確保多個(gè)線程在訪問(wèn)共享資源時(shí)能夠正確地協(xié)調(diào)。Tokio 是一個(gè)強(qiáng)大的異步運(yùn)行時(shí)庫(kù)，為 Rust 提供了多種線程同步機(jī)制。以下是一些常見的同步機(jī)制：

Mutex
RwLock
Barrier
Semaphore
Notify
oneshot 和 mpsc 通道
watch 通道

1. Mutex

Mutex（互斥鎖）是最常見的同步原語(yǔ)之一，用于保護(hù)共享數(shù)據(jù)。它確保同一時(shí)間只有一個(gè)線程能夠訪問(wèn)數(shù)據(jù)，從而避免競(jìng)爭(zhēng)條件。

use tokio::sync::Mutex;
use std::sync::Arc;
?
#[tokio::main]
async fn main() {
    let data = Arc::new(Mutex::new(0));
?
    let mut handles = vec![];
    for _ in 0..10 {
        let data = data.clone();
        let handle = tokio::spawn(async move {
            let mut lock = data.lock().await;
            *lock += 1;
        });
        handles.push(handle);
    }
?
    for handle in handles {
        handle.await.unwrap();
    }
?
    println!("Result: {}", *data.lock().await);
}

2. RwLock

RwLock（讀寫鎖）允許多線程同時(shí)讀取數(shù)據(jù)，但只允許一個(gè)線程寫入數(shù)據(jù)。它比 Mutex 更加靈活，因?yàn)樵谧x取多于寫入的場(chǎng)景下，它能提高性能。功能上，他是讀寫互斥、寫寫互斥、讀讀兼容。

use tokio::sync::RwLock;
use std::sync::Arc;
?
#[tokio::main]
async fn main() {
    let data = Arc::new(RwLock::new(0));
?
    let read_data = data.clone();
    let read_handle = tokio::spawn(async move {
        let lock = read_data.read().await;
        println!("Read: {}", *lock);
    });
?
    let write_data = data.clone();
    let write_handle = tokio::spawn(async move {
        let mut lock = write_data.write().await;
        *lock += 1;
        println!("Write: {}", *lock);
    });
?
    read_handle.await.unwrap();
    write_handle.await.unwrap();
}

3. Barrier

Barrier 是一種同步機(jī)制，允許多個(gè)線程在某個(gè)點(diǎn)上進(jìn)行同步。當(dāng)線程到達(dá)屏障時(shí)，它們會(huì)等待直到所有線程都到達(dá)，然后一起繼續(xù)執(zhí)行。

use tokio::sync::Barrier;
use std::sync::Arc;
?
#[tokio::main]
async fn main() {
    let barrier = Arc::new(Barrier::new(3));
?
    let mut handles = vec![];
    for i in 0..3 {
        let barrier = barrier.clone();
        let handle = tokio::spawn(async move {
            println!("Before wait: {}", i);
            barrier.wait().await;
            println!("After wait: {}", i);
        });
        handles.push(handle);
    }
?
    for handle in handles {
        handle.await.unwrap();
    }
}

4. Semaphore

Semaphore（信號(hào)量）是一種用于控制對(duì)資源訪問(wèn)的同步原語(yǔ)。它允許多個(gè)線程訪問(wèn)資源，但有一個(gè)最大并發(fā)數(shù)限制。

#[tokio::test]
async fn test_sem() {
    let semaphore = Arc::new(Semaphore::new(3));
?
    let mut handles = vec![];
    for i in 0..5 {
        let semaphore = semaphore.clone();
        let handle = tokio::spawn(async move {
            let permit = semaphore.acquire().await.unwrap();
            let now = Local::now();
            println!("Got permit: {} at {:?}", i, now);
            println!(
                "Semaphore available permits before sleep: {}",
                semaphore.available_permits()
            );
            sleep(Duration::from_secs(5)).await;
            drop(permit);
            println!(
                "Semaphore available permits after sleep: {}",
                semaphore.available_permits()
            );
        });
        handles.push(handle);
    }
?
    for handle in handles {
        handle.await.unwrap();
    }
}

最終的結(jié)果如下

Got permit: 0 at 2024-08-08T21:03:04.374666+08:00
Semaphore available permits before sleep: 2
Got permit: 1 at 2024-08-08T21:03:04.375527800+08:00
Semaphore available permits before sleep: 1
Got permit: 2 at 2024-08-08T21:03:04.375563+08:00
Semaphore available permits before sleep: 0
Semaphore available permits after sleep: 0
Semaphore available permits after sleep: 0
Semaphore available permits after sleep: 1
Got permit: 3 at 2024-08-08T21:03:09.376722800+08:00
Semaphore available permits before sleep: 1
Got permit: 4 at 2024-08-08T21:03:09.376779200+08:00
Semaphore available permits before sleep: 1
Semaphore available permits after sleep: 2
Semaphore available permits after sleep: 3

5. Notify

Notify 是一種用于線程間通知的簡(jiǎn)單機(jī)制。它允許一個(gè)線程通知其他線程某些事件的發(fā)生。

use tokio::sync::Notify;
use std::sync::Arc;
?
#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let notify_clone = notify.clone();
?
    let handle = tokio::spawn(async move {
        notify_clone.notified().await;
        println!("Received notification");
    });
?
    notify.notify_one();
    handle.await.unwrap();
}

6. oneshot 和 mpsc 通道

oneshot 通道用于一次性發(fā)送消息，而 mpsc 通道則允許多個(gè)生產(chǎn)者發(fā)送消息到一個(gè)消費(fèi)者。一般地onshot用于異常通知、啟動(dòng)分析等功能。mpsc用于實(shí)現(xiàn)異步消息同步

oneshot

use tokio::sync::oneshot;
?
#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel();
?
    tokio::spawn(async move {
        tx.send("Hello, world!").unwrap();
    });
?
    let message = rx.await.unwrap();
    println!("Received: {}", message);
}

mpsc

use tokio::sync::mpsc;
?
#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
?
    tokio::spawn(async move {
        tx.send("Hello, world!").await.unwrap();
    });
?
    while let Some(message) = rx.recv().await {
        println!("Received: {}", message);
    }
}

7. watch 通道

watch 通道用于發(fā)送和接收共享狀態(tài)的更新。它允許多個(gè)消費(fèi)者監(jiān)聽狀態(tài)的變化。

use tokio::sync::watch;
?
#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel("initial");
?
    tokio::spawn(async move {
        tx.send("updated").unwrap();
    });
?
    while rx.changed().await.is_ok() {
        println!("Received: {}", *rx.borrow());
    }
}

?watch通道?：