Rust 通過異步實現(xiàn)并發(fā)的方法示例
在本文中，我們將重點討論線程和 future 之間的區(qū)別。
在許多情況下，使用異步處理并發(fā)性的 API 與使用線程的 API 非常相似，但它們通常具有不同的行為，并且它們幾乎總是具有不同的性能特征。
用 spawn_task 創(chuàng)建一個新任務(wù)
使用 thread::spawn 可以創(chuàng)建一個新線程，我們編寫的第一個程序是在兩個單獨的線程上進行計數(shù)。
讓我們使用 async 做同樣的事情。trpl crate 提供了一個看起來與 thread::spawn API 非常相似的 spawn_task 函數(shù)，以及一個 sleep 函數(shù)，它是 thread::sleep API 的異步版本。我們可以一起使用它們來實現(xiàn)計數(shù)示例：
use std::time::Duration;
fn main() {
trpl::run(async {
trpl::spawn_task(async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
}
我們使用 trpl::run 設(shè)置 main 函數(shù)，以便我們的頂級函數(shù)可以是異步的。
然后我們在該塊中編寫兩個循環(huán)，每個循環(huán)都包含一個 trpl::sleep 調(diào)用，該調(diào)用在發(fā)送下一條消息之前等待半秒。我們將一個循環(huán)放在 trpl::spawn_task 中，另一個放在頂層的 for 循環(huán)中。我們還在 sleep 調(diào)用之后添加了一個 await。
這段代碼的行為類似于基于線程的實現(xiàn)——包括當你運行它時，你可能會看到消息以不同的順序出現(xiàn)在你自己的終端上。
這個版本在主異步塊體中的 for 循環(huán)完成后立即停止，因為在主函數(shù)結(jié)束時，由 spawn_task 生成的任務(wù)被關(guān)閉。如果希望它一直運行到任務(wù)完成，則需要使用連接句柄來等待第一個任務(wù)完成。對于線程，我們使用 join 方法來“阻塞”，直到線程完成運行。我們可以使用 await 來做同樣的事情，因為任務(wù)句柄本身就是一個 future。它的輸出類型是 Result，所以我們也在等待它之后展開它。
use std::time::Duration;
fn main() {
trpl::run(async {
let handle = trpl::spawn_task(async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
handle.await.unwrap();
});
}
這個更新的版本運行直到兩個循環(huán)結(jié)束。
到目前為止，看起來 async 和線程給出了相同的基本結(jié)果，只是語法不同：使用 await 而不是在連接句柄上調(diào)用 join，并等待 sleep 調(diào)用。
更大的區(qū)別在于，我們不需要生成另一個操作系統(tǒng)線程來執(zhí)行此操作。實際上，我們甚至不需要在這里生成任務(wù)。由于 async 塊編譯為匿名的 future，我們可以將每個循環(huán)放在 async 塊中，并讓運行時使用 trpl::join 函數(shù)將它們運行到完成。
我們之前展示了如何在調(diào)用 std::thread::spawn 時對返回的 JoinHandle 類型使用 join 方法。trpl::join 函數(shù)與此類似，但用于 future。當你給它兩個 future 時，它會產(chǎn)生一個新的 future，它的輸出是一個元組，其中包含你傳入的每個 future 完成后的輸出。
我們使用 trpl::join 來等待 fut1 和 fut2 完成。我們不等待 fut1 和 fut2，而是等待 trpl::join 生成的新 future。
use std::time::Duration;
fn main() {
trpl::run(async {
let fut1 = async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
};
let fut2 = async {
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
};
trpl::join(fut1, fut2).await;
});
}
編譯運行，我們看到兩個 future 都運行到完成：
現(xiàn)在，每次運行的結(jié)果的順序都完全相同，這與我們在線程中看到的非常不同。
這是因為 trpl::join 函數(shù)是公平的，這意味著它同樣頻繁地檢查每個 future，在它們之間交替，如果另一個準備好了，它永遠不會讓一個搶先。對于線程，操作系統(tǒng)決定檢查哪個線程以及讓它運行多長時間。對于異步 Rust，運行時決定檢查哪個任務(wù)。
在實踐中，細節(jié)變得復(fù)雜，因為異步運行時可能會在后臺使用操作系統(tǒng)線程作為管理并發(fā)性的一部分，因此保證公平性對運行時來說可能需要更多的工作。
運行時不必保證任何給定操作的公平性，它們通常提供不同的 API，讓你選擇是否需要公平性。
使用消息傳遞計算兩個任務(wù)
我們使用消息傳遞的異步版本在 future 之間共享數(shù)據(jù)。
我們將采用與使用消息傳遞在線程之間傳輸數(shù)據(jù)略有不同的方法來說明基于線程的并發(fā)和基于 future 的并發(fā)之間的一些關(guān)鍵區(qū)別。
在 trpl::run 的 async 塊中創(chuàng)建通道：
fn main() {
trpl::run(async {
let (tx, mut rx) = trpl::channel();
let val = String::from("hi");
tx.send(val).unwrap();
let received = rx.recv().await.unwrap();
println!("Got: {received}");
});
}
這里，我們使用 trpl::channel，這是 std::mpsc::channel（多生產(chǎn)者、單消費者通道）的異步版本。異步版本的 API 與基于線程的版本只有一點不同：它使用一個可變的接收端 rx，它的 recv 方法產(chǎn)生一個我們需要等待的 future，而不是直接產(chǎn)生值?，F(xiàn)在我們可以將消息從發(fā)送者發(fā)送到接收者。注意，我們不需要生成一個單獨的線程或任務(wù)，我們只需要等待 rx.recv 調(diào)用。
在 std::mpsc::channel 中的 Receiver::recv 方法阻塞線程，直到它接收到消息。trpl::Receiver::recv 方法是異步的，它不阻塞，而是將控制權(quán)交還給運行時，直到接收到消息或通道的發(fā)送端關(guān)閉為止。相比之下，我們不等待 send 調(diào)用，因為它不會阻塞。
注意：由于所有這些異步代碼都在 trpl::run 調(diào)用中的異步塊中運行，因此其中的所有代碼都可以避免阻塞。但是，它外面的代碼將在運行函數(shù)返回時阻塞。這就是 trpl::run 函數(shù)的全部意義：它允許你選擇在哪里阻塞某些異步代碼集，以及在哪里在同步代碼和異步代碼之間轉(zhuǎn)換。在大多數(shù)異步運行時，run 實際上被命名為 block_on 正是出于這個原因。
關(guān)于這個例子，請注意兩點。首先，消息會馬上到達。第二，雖然我們在這里使用了 future，但是還沒有并發(fā)。程序的一切都是按順序進行的，就像不涉及 future 一樣。
讓我們通過發(fā)送一系列消息并在它們之間休眠來解決第一部分：
use std::time::Duration;
fn main() {
trpl::run(async {
let (tx, mut rx) = trpl::channel();
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
});
}
Rust 還沒有一種方法可以在一系列異步項上編寫 for 循環(huán)，因此我們需要使用 while let 條件循環(huán)，只要循環(huán)指定的模式繼續(xù)匹配該值，循環(huán)就會繼續(xù)執(zhí)行。
rx.recv() 產(chǎn)生一個我們等待的 future。運行時將暫停 future，直到它準備好。一旦消息到達，future 將解析為 Some(message)。當通道關(guān)閉時，無論是否有消息到達，future 都將解析為 None，表示沒有更多的值，因此我們應(yīng)該停止輪詢——也就是說，停止 await。
while let 循環(huán)將所有這些組合在一起。如果調(diào)用 rx.recv().await 的結(jié)果是Some(message)，則可以訪問該消息，并可以在循環(huán)體中使用它。如果結(jié)果為 None，則循環(huán)結(jié)束。每次循環(huán)完成時，它都會再次到達等待點，因此運行時將再次暫停它，直到另一條消息到達。
代碼現(xiàn)在成功地發(fā)送和接收了所有消息：
不幸的是，仍然存在一些問題。首先，消息不會以半秒的間隔到達，它們在我們啟動程序后 2 秒同時到達。其次，這個程序永遠不會退出！相反，它會永遠等待新的消息。
因為程序中只有一個異步塊，因此其中的所有內(nèi)容都是線性運行的，仍然沒有并發(fā)性。所有的 tx.send 調(diào)用都會發(fā)生，并與所有的 trpl::sleep 調(diào)用及其相關(guān)的等待點穿插在一起。只有這樣，while let 循環(huán)才能通過 recv 調(diào)用上的任何等待點。
為了獲得我們想要的行為，即在每個消息之間發(fā)生睡眠延遲，我們需要將 tx 和 rx 操作放在各自的異步塊中，然后運行時可以使用 trpl::join 分別執(zhí)行它們中的每一個。同樣，我們等待調(diào)用 trpl::join 的結(jié)果，而不是單個的 future。
use std::time::Duration;
fn main() {
trpl::run(async {
let (tx, mut rx) = trpl::channel();
let tx_fut = async {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
trpl::join(tx_fut, rx_fut).await;
});
}
消息以 500 ms 的間隔打印，而不是在 2 s 后匆忙打印。
然而，由于 while let 循環(huán)與 trpl::join 的交互方式，程序仍然不會退出：
- 只有當傳遞給它的兩個 future 都完成后，從 trpl::join 返回的 future 才會完成。
- 在發(fā)送 vals 中的最后一條消息后，一旦結(jié)束 sleep，tx future 就完成了。
- 直到 while let 循環(huán)結(jié)束，rx future 才會完成。
- while let 循環(huán)直到等待 rx.recv 產(chǎn)生 None 才會結(jié)束。
- 等待 rx.recv 只會在通道的另一端關(guān)閉時返回 None。
- 只有當我們調(diào)用 rx.close 或當發(fā)送端 tx 被丟棄時，通道才會關(guān)閉。
- 我們不會在任何地方調(diào)用 rx.close，并且在傳遞給 trpl::run 的最外層異步塊結(jié)束之前，tx 不會被丟棄。
- 這個塊不能結(jié)束，因為它在 trpl::join 完成時被阻塞了，這將我們帶回到列表的頂部。
我們可以通過在某處調(diào)用 rx.close 來手動關(guān)閉 rx，但這沒有多大意義。在處理任意數(shù)量的消息后停止將使程序關(guān)閉，但我們可能會錯過消息。我們需要一些其他的方法來確保 tx 在函數(shù)結(jié)束前被刪除。
現(xiàn)在，我們發(fā)送消息的異步塊只借用 tx，因為發(fā)送消息不需要所有權(quán)，但是如果我們可以將 tx 移動到異步塊中，那么一旦該塊結(jié)束，它就會被丟棄。move 關(guān)鍵字對異步塊的作用就像對閉包的作用一樣，將數(shù)據(jù)轉(zhuǎn)移到異步塊中。
我們將用于發(fā)送消息的塊從 async 更改為 async move。當我們運行這個版本的代碼時，它會在發(fā)送和接收最后一條消息后優(yōu)雅地關(guān)閉。
use std::time::Duration;
fn main() {
trpl::run(async {
let (tx, mut rx) = trpl::channel();
let tx_fut = async move {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
trpl::join(tx_fut, rx_fut).await;
});
}
因為 tx 所有權(quán)被轉(zhuǎn)移到 async 塊內(nèi)，在該塊執(zhí)行完也就是發(fā)送作業(yè)結(jié)束之后，tx 隨之被銷毀，觸發(fā)通道關(guān)閉，接收端返回 None。
這個異步通道也是一個多生產(chǎn)者通道，所以如果我們想從多個 future 發(fā)送消息，我們可以在 tx 上調(diào)用 clone。
use std::time::Duration;
fn main() {
trpl::run(async {
let (tx, mut rx) = trpl::channel();
let tx1 = tx.clone();
let tx1_fut = async move {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx1.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
let tx_fut = async move {
let vals = vec![
String::from("more"),
String::from("messages"),
String::from("for"),
String::from("you"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(1500)).await;
}
};
trpl::join3(tx1_fut, tx_fut, rx_fut).await;
});
}
克隆 tx，在第一個異步塊之外創(chuàng)建 tx1，我們將 tx1 移動到該塊中。然后將原始 tx 移動到一個新的異步塊中，在那里我們以稍慢的延遲發(fā)送更多消息。
用于發(fā)送消息的兩個異步塊都需要是 async move 塊，以便在這些塊完成時丟棄 tx 和 tx1。最后，我們從 trpl::join 切換到 trpl::join3 來處理額外的 future。
現(xiàn)在我們看到了來自兩個發(fā)送 future 的所有消息，由于發(fā)送 future 在發(fā)送后使用的延遲略有不同，因此接收消息的間隔也不同。
這是一個良好的開端，但它限制了我們的 future 數(shù)量：兩個對應(yīng) join，或三個對應(yīng) join3。
到此這篇關(guān)于Rust 通過異步實現(xiàn)并發(fā)的方法示例的文章就介紹到這了,更多相關(guān)Rust 異步并發(fā)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Rust實現(xiàn)grep命令行工具的方法，本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-07-07
這篇文章主要為大家詳細介紹了如何利用Rust語言實現(xiàn)AES加解密算法，文中的示例代碼講解詳細，具有一定的借鑒價值，需要的可以了解一下2022-10-10
在Rust中要用Struct和Enum組織數(shù)據(jù)的原因解析
在Rust中,Struct和Enum是組織數(shù)據(jù)的核心工具,Struct用于將相關(guān)字段封裝為單一實體,便于管理和擴展,Enum用于明確定義所有可能的狀態(tài),本文將通過具體示例,深入探討為什么在Rust中必須使用struct和enum來管理數(shù)據(jù),感興趣的朋友一起學(xué)習(xí)吧2025-02-02
Rust中的模式匹配提供了多種方式來處理不同的數(shù)據(jù)類型和場景,本文給大家介紹Rust 中的模式匹配語法,感興趣的朋友一起看看吧2025-03-03
這篇文章主要為大家詳細介紹了Rust如何在線程的閉包中安全的使用變量,包括共享變量和修改變量,文中的示例代碼講解詳細,有需要的小伙伴可以參考下2025-01-01
如何使用bindgen將Ｃ語言頭文件轉(zhuǎn)換為Rust接口代碼
這篇文章主要介紹了使用bindgen將Ｃ語言頭文件轉(zhuǎn)換為Rust接口代碼,本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-01-01
這篇文章主要介紹了Rust中FFI編程知識點整理總結(jié),本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-09-09
這篇文章主要介紹了用rust?寫一個jar包?class沖突檢測工具?的相關(guān)資料,需要的朋友可以參考下2023-05-05
Substrate是一個用于構(gòu)建區(qū)塊鏈的開發(fā)框架，它由Parity團隊基于Rust語言開發(fā)而成，是一個開箱即用的區(qū)塊鏈構(gòu)造器，本文詳細介紹了Substrate框架中的Core，需要的朋友可以參考下2023-05-05
這篇文章主要介紹了Rust語言的println!宏的格式占位符,這只是格式說明符的一部分清單,Rust?的格式化系統(tǒng)非常靈活和強大,支持更多的選項和組合,需要的朋友可以參考下2024-03-03